Clean up temporary files created by forceToDiskExecution #1621

ajohnson-stripe-zz · 2016-11-14T19:18:48Z

This adds functionality for WriteExecution to keep track of temporary files created and then delete them after execution has finished. Currently this is only used by forceToDiskExecution, which previously would leave around temporary files every time it was used.

This fixes #1615.

oscar-stripe

This is great! Thanks for attacking this problem.

oscar-stripe · 2016-11-14T19:26:20Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

          val newFn = { (conf: Config, mode: Mode) =>
            (fn(conf, mode), otherFn(conf, mode))
          }
-          WriteExecution(head, h :: t ::: tail, newFn)
+          WriteExecution(head, h :: t ::: tail, newFn, tempFiles)


don't you need to merge the sets from the left and right? Seems like we are losing this.tempFilesToCleanup.

Ah good catch, I do need that here.

oscar-stripe · 2016-11-14T19:29:34Z

scalding-core/src/test/scala/com/twitter/scalding/ExecutionTest.scala

+      Set(tempFile)
+    }
+
+    Execution.write(writeFn, readFn, filesToDeleteFn)


can you test a .zip to be sure we don't lose files there?

oscar-stripe · 2016-11-14T19:30:24Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+        try {
+          val path = new Path(file)
+          if (fs.exists(path)) {
+            fs.delete(path, true)


can you comment what the boolean means at this call site?

oscar-stripe · 2016-11-14T19:30:48Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+        case hdfsMode: HadoopMode =>  FileSystem.get(hdfsMode.jobConf)
+      }
+
+      filesToCleanup foreach { file: String =>


can we prefer the . style and use filesToCleanup.foreach {

ajohnson-stripe-zz · 2016-11-14T21:55:24Z

Thanks for the feedback @oscar-stripe. I incorporated in the changes you suggested.

I also noticed that the REPL expects that the temporary outputs from forceToDiskExecution stick around in things like the snapshot method on ShellTypedPipe. As such I moved the actual deletion to happen in a shutdown hook on JVM exit.

ajohnson-stripe-zz · 2016-11-14T22:28:56Z

Hmm, I'm not sure I see why the CI build is failing. It's complaining about failing when compiling maple: .[error] (maple/compile:compileIncremental) javac returned nonzero exit code, but I only see warnings from the compiler, no errors. I can build that fine locally too.

I do see this in the log, maybe related?

.tar: Unexpected EOF in archive
tar: rmtlseek not stopped at a record boundary
tar: Error is not recoverable: exiting now

piyushnarang · 2016-11-14T22:33:03Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+          }
+        } catch {
+          // If we fail in deleting a temp file, log the error but don't fail the run
+          case e: Exception => LOG.info(s"Unable to delete temp file $file")


can we pass e to the log statement? that way we get the message and stacktrace? Also should this be a warn?

piyushnarang · 2016-11-14T22:34:56Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

@@ -284,6 +288,34 @@ object Execution {
  }

  /**
+    * This is a Thread used as a shutdown hook to clean up temporary files created by some Execution
+    */
+  case class TempFileCleanup(filesToCleanup: mutable.Set[String], mode: Mode) extends Thread {


since we're just iterating over the filesToCleanup in this case class, maybe we can make this a more generic interface rather than mutable.Set? Iterable / Seq / Set?

piyushnarang · 2016-11-14T22:44:23Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+            // The "true" parameter here indicates that we should recursively delete everything under the given path
+            fs.delete(path, true)
+          }
+        } catch {


as this is part of a shutdown hook, should we add a catch all clause?

Do you mean catch { case e: Throwable => so we catch anything? Is that what you mean?

@oscar-stripe yeah. If we for whatever reason throw an caught exception in this code that will result in the shutdown code not being run completely as I understand it.

piyushnarang · 2016-11-14T22:47:21Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

@@ -284,6 +288,34 @@ object Execution {
  }

  /**
+    * This is a Thread used as a shutdown hook to clean up temporary files created by some Execution
+    */
+  case class TempFileCleanup(filesToCleanup: mutable.Set[String], mode: Mode) extends Thread {


maybe also call out that if the job is aborted this will not cleanup stuff (if folks haven't read / know about shutdown hook semantics)

oscar-stripe · 2016-11-15T00:51:56Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

@@ -284,6 +288,34 @@ object Execution {
  }

  /**
+    * This is a Thread used as a shutdown hook to clean up temporary files created by some Execution
+    */
+  case class TempFileCleanup(filesToCleanup: mutable.Set[String], mode: Mode) extends Thread {


+1 to taking an immutable Iterable here. Can we make the class private?

oscar-stripe · 2016-11-15T00:54:12Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+            // The "true" parameter here indicates that we should recursively delete everything under the given path
+            fs.delete(path, true)
+          }
+        } catch {


Do you mean catch { case e: Throwable => so we catch anything? Is that what you mean?

ajohnson-stripe-zz · 2016-11-15T16:17:15Z

I've incorporated the next round of feedback. I'm still looking into it, but I'm not able to reproduce the CI build failure.

johnynek · 2016-11-15T16:23:16Z

I restarted. Let's see if it is a CI issue.

On Tue, Nov 15, 2016 at 8:17 AM Andrew Johnson notifications@github.com
wrote:

I've incorporated the next round of feedback. I'm still looking into it,
but I'm not able to reproduce the CI build failure.

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
#1621 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAEJdvQAZ-YG1YeGlmmZu_ynvCm-F0h0ks5q-dsLgaJpZM4Kxsnr
.

This reverts commit b9a04b6.

piyushnarang · 2016-11-15T18:46:18Z

@ajohnson-stripe changes look good to me. 👍

johnynek · 2016-11-15T18:49:15Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

@@ -313,7 +347,7 @@ object Execution {
      new EvalCache {
        override protected[EvalCache] val messageQueue: LinkedBlockingQueue[EvalCache.FlowDefAction] = self.messageQueue
        override def start(): Unit = sys.error("Invalid to start child EvalCache")
-        override def finished(): Unit = sys.error("Invalid to finish child EvalCache")
+        override def finished(mode: Mode): Unit = sys.error("Invalid to finish child EvalCache")


don't we need to override the addFilesToCleanup to forward back to the parent list? I think any files added to the child will not be deleted, right?

Yeah, that does need to be added

johnynek · 2016-11-15T19:25:09Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

@@ -305,15 +338,17 @@ object Execution {
    type Counters = Map[Long, ExecutionCounters]
    private[this] val cache = new FutureCache[(Config, Execution[Any]), (Any, Counters)]
    private[this] val toWriteCache = new FutureCache[(Config, ToWrite), Counters]
+    protected[EvalCache] val filesToCleanup = mutable.Set[String]()


actually, I think we should keep this private and override the add method. The reason is we forgot one more thing: this is not thread-safe, but we have so many threads in play, we should make this thread-safe.

I think we are modifying this value from several threads.

So, if we just forward the method, we can handle the locking a bit more cleanly inside that method.

Good point, I'll switch that up.

isnotinvain · 2016-11-15T19:57:20Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+          }
+        } catch {
+          // If we fail in deleting a temp file, log the error but don't fail the run
+          case e: Throwable => LOG.warn(s"Unable to delete temp file $file", e)


style-wise I think case e => LOG.warn(s"Unable to delete temp file $file", e) is the same (without the :Throwable part)

It is the same, but leaving out the : Throwable generates a compiler warning requesting that you be explicit. I see a mix of both in the codebase now, but I'm happy to go with whatever's the preferred style.

isnotinvain · 2016-11-15T20:01:52Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+    *
+    * If the job is aborted the shutdown hook may not run and the temporary files will not get cleaned up
+    */
+  private[scalding] case class TempFileCleanup(filesToCleanup: Iterable[String], mode: Mode) extends Thread {


should this extend Runnable instead? Does it have to extend Thread?

addShutdownHook will only take a Thread unfortunately

johnynek · 2016-11-15T20:03:37Z

👍 looks good to me.

johnynek · 2016-11-16T05:49:21Z

For throwable we should be explicit and avoid the warning in my view.
On Tue, Nov 15, 2016 at 12:05 Andrew Johnson notifications@github.com
wrote:

@ajohnson-stripe commented on this pull request.

In scalding-core/src/main/scala/com/twitter/scalding/Execution.scala
#1621:
override def run(): Unit = {
 val fs = mode match {
   case localMode: CascadingLocal => FileSystem.getLocal(new Configuration)
   case hdfsMode: HadoopMode =>  FileSystem.get(hdfsMode.jobConf)
 }
 filesToCleanup.foreach { file: String =>
   try {
     val path = new Path(file)
     if (fs.exists(path)) {
       // The "true" parameter here indicates that we should recursively delete everything under the given path
       fs.delete(path, true)
     }
   } catch {
     // If we fail in deleting a temp file, log the error but don't fail the run
     case e: Throwable => LOG.warn(s"Unable to delete temp file $file", e)
It is the same, but leaving out the : Throwable generates a compiler
warning requesting that you be explicit. I see a mix of both in the
codebase now, but I'm happy to go with whatever's the preferred style.

—
You are receiving this because you commented.

Reply to this email directly, view it on GitHub
#1621, or mute the thread
https://github.com/notifications/unsubscribe-auth/AAEJdpRzfsyQs4txuM5bz7H4UcdXfz_9ks5q-hCHgaJpZM4Kxsnr
.

oscar-stripe suggested changes Nov 14, 2016

View reviewed changes

ajohnson-stripe-zz force-pushed the forceToDiskExecution-cleanup branch 2 times, most recently from f8a6cf2 to 53de7ff Compare November 14, 2016 21:53

piyushnarang reviewed Nov 14, 2016

View reviewed changes

oscar-stripe reviewed Nov 15, 2016

View reviewed changes

ajohnson-stripe-zz force-pushed the forceToDiskExecution-cleanup branch from 53de7ff to b34d3d4 Compare November 15, 2016 12:56

Clean up temporary files created by forceToDiskExecution

b34d3d4

ajohnson-stripe added 2 commits November 15, 2016 11:53

Try disabling travis caching

b9a04b6

Revert "Try disabling travis caching"

953c96e

This reverts commit b9a04b6.

johnynek reviewed Nov 15, 2016

View reviewed changes

ajohnson-stripe added 2 commits November 15, 2016 13:54

Forward filesToCleanup to parent in cleanCache

8903162

consistency

f28d7e5

johnynek reviewed Nov 15, 2016

View reviewed changes

synchronize in addFilesToCleanup

6ac83e8

isnotinvain reviewed Nov 15, 2016

View reviewed changes

johnynek merged commit d389c63 into twitter:develop Nov 17, 2016

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Clean up temporary files created by forceToDiskExecution #1621

Clean up temporary files created by forceToDiskExecution #1621

ajohnson-stripe-zz commented Nov 14, 2016

oscar-stripe left a comment

oscar-stripe Nov 14, 2016

ajohnson-stripe-zz Nov 14, 2016

oscar-stripe Nov 14, 2016

oscar-stripe Nov 14, 2016

ajohnson-stripe-zz Nov 14, 2016

oscar-stripe Nov 14, 2016

ajohnson-stripe-zz Nov 14, 2016

ajohnson-stripe-zz commented Nov 14, 2016

ajohnson-stripe-zz commented Nov 14, 2016

piyushnarang Nov 14, 2016

piyushnarang Nov 14, 2016

ajohnson-stripe-zz Nov 15, 2016

piyushnarang Nov 14, 2016

oscar-stripe Nov 15, 2016

piyushnarang Nov 15, 2016

piyushnarang Nov 14, 2016

oscar-stripe Nov 15, 2016

oscar-stripe Nov 15, 2016

ajohnson-stripe-zz commented Nov 15, 2016

johnynek commented Nov 15, 2016

piyushnarang commented Nov 15, 2016

johnynek Nov 15, 2016

ajohnson-stripe-zz Nov 15, 2016

johnynek Nov 15, 2016

ajohnson-stripe-zz Nov 15, 2016

isnotinvain Nov 15, 2016

ajohnson-stripe-zz Nov 15, 2016

isnotinvain Nov 15, 2016

ajohnson-stripe-zz Nov 15, 2016

johnynek commented Nov 15, 2016

johnynek commented Nov 16, 2016

@ajohnson-stripe commented on this pull request.

Clean up temporary files created by forceToDiskExecution #1621

Clean up temporary files created by forceToDiskExecution #1621

Conversation

ajohnson-stripe-zz commented Nov 14, 2016

oscar-stripe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajohnson-stripe-zz commented Nov 14, 2016

ajohnson-stripe-zz commented Nov 14, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajohnson-stripe-zz commented Nov 15, 2016

johnynek commented Nov 15, 2016

piyushnarang commented Nov 15, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johnynek commented Nov 15, 2016

johnynek commented Nov 16, 2016

@ajohnson-stripe commented on this pull request.