Worker.executeAfter() API for scheduling jobs on worker. (#2971)

backend.native/tests/build.gradle

@@ -762,7 +762,7 @@ task worker8(type: RunKonanTest) {
 
 task worker9(type: RunKonanTest) {
     disabled = (project.testTarget == 'wasm32') // Workers need pthreads.
-    goldValue = "zzz\n42\nOK\n"
+    goldValue = "zzz\n42\nOK\nfirst 2\nsecond 3\nfrozen OK\n"
     source = "runtime/workers/worker9.kt"
 }
 

backend.native/tests/runtime/workers/worker9.kt

@@ -9,7 +9,7 @@ import kotlin.test.*
 
 import kotlin.native.concurrent.*
 
-@Test fun runTest() {
+@Test fun runTest1() {
     withLock { println("zzz") }
     val worker = Worker.start()
     val future = worker.execute(TransferMode.SAFE, {}) {
@@ -25,3 +25,38 @@ import kotlin.native.concurrent.*
 fun withLock(op: () -> Unit) {
     op()
 }
+
+@Test fun runTest2() {
+    val worker = Worker.start()
+    val future = worker.execute(TransferMode.SAFE, {}) {
+        val me = Worker.current!!
+        var x = 1
+        me.executeAfter (20000) {
+            println("second ${++x}")
+        }
+        me.executeAfter(10000) {
+            println("first ${++x}")
+        }
+    }
+    worker.requestTermination().result
+}
+
+@Test fun runTest3() {
+    val worker = Worker.start()
+    assertFailsWith<IllegalStateException> {
+        worker.executeAfter {
+            println("shall not happen")
+        }
+    }
+    assertFailsWith<IllegalArgumentException> {
+        worker.executeAfter(-1, {
+            println("shall not happen")
+        }.freeze())
+    }
+
+    worker.executeAfter(0, {
+        println("frozen OK")
+    }.freeze())
+
+    worker.requestTermination().result
+}

runtime/src/main/cpp/Types.h

@@ -26,6 +26,7 @@
 
 #include <deque>
 #include <string>
+#include <set>
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
@@ -69,6 +70,8 @@ template<class Value>
 using KStdUnorderedSet = std::unordered_set<Value,
   std::hash<Value>, std::equal_to<Value>,
   KonanAllocator<Value>>;
+template<class Value, class Compare>
+using KStdOrderedSet = std::set<Value, Compare, KonanAllocator<Value>>;
 template<class Value>
 using KStdVector = std::vector<Value, KonanAllocator<Value>>;
 

runtime/src/main/cpp/Worker.cpp

@@ -59,6 +59,12 @@ enum {
   UNCHECKED = 1
 };
 
+enum JobKind {
+  JOB_REGULAR = 1,
+  JOB_TERMINATE,
+  JOB_EXECUTE_AFTER
+};
+
 THREAD_LOCAL_VARIABLE KInt g_currentWorkerId = 0;
 
 KNativePtr transfer(ObjHolder* holder, KInt mode) {
@@ -92,10 +98,20 @@ class Future {
   }
 
   ~Future() {
+    clear();
     pthread_mutex_destroy(&lock_);
     pthread_cond_destroy(&cond_);
   }
 
+  void clear() {
+    Locker locker(&lock_);
+    if (result_ != nullptr) {
+      // No one cared to consume result - dispose it.
+      DisposeStablePointer(result_);
+      result_ = nullptr;
+    }
+  }
+
   OBJ_GETTER0(consumeResultUnlocked) {
     Locker locker(&lock_);
     while (state_ == SCHEDULED) {
@@ -130,12 +146,36 @@ class Future {
 };
 
 struct Job {
-  KRef (*function)(KRef, ObjHeader**);
+  enum JobKind kind;
-  KNativePtr argument;
+  union {
-  Future* future;
+    struct {
-  KInt transferMode;
+      KRef (*function)(KRef, ObjHeader**);
+      KNativePtr argument;
+      Future* future;
+      KInt transferMode;
+    } regularJob;
+
+    struct {
+      Future* future;
+      bool waitDelayed;
+    } terminationRequest;
+
+    struct {
+      KNativePtr operation;
+      KLong whenExecute;
+    } executeAfter;
+  };
 };
 
+struct JobCompare {
+  bool operator() (const Job& lhs, const Job& rhs) const {
+    RuntimeAssert(lhs.kind == JOB_EXECUTE_AFTER && rhs.kind == JOB_EXECUTE_AFTER, "Must be delayed jobs");
+    return lhs.executeAfter.whenExecute < rhs.executeAfter.whenExecute;
+  }
+};
+
+typedef KStdOrderedSet<Job, JobCompare> DelayedJobSet;
+
 class Worker {
  public:
   Worker(KInt id, bool errorReporting) : id_(id), errorReporting_(errorReporting) {
@@ -144,10 +184,29 @@ class Worker {
   }
 
   ~Worker() {
-    // Cleanup jobs in queue.
+    // Cleanup jobs in the queue.
     for (auto job : queue_) {
-      DisposeStablePointer(job.argument);
+      switch (job.kind) {
-      job.future->cancelUnlocked();
+        case JOB_REGULAR:
+          DisposeStablePointer(job.regularJob.argument);
+          job.regularJob.future->cancelUnlocked();
+          break;
+        case JOB_EXECUTE_AFTER: {
+          // TODO: what do we do here? Shall we execute them?
+          DisposeStablePointer(job.executeAfter.operation);
+          break;
+        }
+        case JOB_TERMINATE: {
+          // TODO: any more processing here?
+          job.terminationRequest.future->cancelUnlocked();
+          break;
+        }
+      }
+    }
+
+    for (auto job : delayed_) {
+       RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed");
+       DisposeStablePointer(job.executeAfter.operation);
     }
 
     pthread_mutex_destroy(&lock_);
@@ -157,22 +216,68 @@ class Worker {
   void putJob(Job job, bool toFront) {
     Locker locker(&lock_);
     if (toFront)
-       queue_.push_front(job);
+      queue_.push_front(job);
     else
-       queue_.push_back(job);
+      queue_.push_back(job);
     pthread_cond_signal(&cond_);
   }
 
+  void putDelayedJob(Job job) {
+    Locker locker(&lock_);
+    delayed_.insert(job);
+    pthread_cond_signal(&cond_);
+  }
+
+  bool waitDelayed() {
+    Locker locker(&lock_);
+    if (delayed_.size() == 0) return false;
+    waitForQueueLocked();
+    return true;
+  }
+
   Job getJob() {
     Locker locker(&lock_);
-    while (queue_.size() == 0) {
+    waitForQueueLocked();
-      pthread_cond_wait(&cond_, &lock_);
-    }
     auto result = queue_.front();
     queue_.pop_front();
     return result;
   }
 
+  KLong checkDelayedLocked() {
+    if (delayed_.size() == 0) {
+      return -1;
+    }
+    auto it = delayed_.begin();
+    auto job = *it;
+    RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed job");
+    auto now = konan::getTimeMicros();
+    if (job.executeAfter.whenExecute <= now) {
+      delayed_.erase(it);
+      queue_.push_back(job);
+      return 0;
+    } else {
+      return job.executeAfter.whenExecute - now;
+    }
+  }
+
+  void waitForQueueLocked() {
+    while (queue_.size() == 0) {
+      KLong closestToRun = checkDelayedLocked();
+      if (closestToRun == 0) continue;
+      if (closestToRun > 0) {
+        struct timeval tv;
+        struct timespec ts;
+        gettimeofday(&tv, nullptr);
+        KLong nsDelta = closestToRun * 1000LL;
+        ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL;
+        ts.tv_sec = (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL;
+        pthread_cond_timedwait(&cond_, &lock_, &ts);
+      } else {
+        pthread_cond_wait(&cond_, &lock_);
+      }
+    }
+  }
+
   KInt id() const { return id_; }
 
   bool errorReporting() const { return errorReporting_; }
@@ -180,9 +285,11 @@ class Worker {
  private:
   KInt id_;
   KStdDeque<Job> queue_;
+  DelayedJobSet delayed_;
   // Lock and condition for waiting on the queue.
   pthread_mutex_t lock_;
   pthread_cond_t cond_;
+  // If errors to be reported on console.
   bool errorReporting_;
 };
 
@@ -222,28 +329,52 @@ class State {
       KInt id, KNativePtr jobFunction, KNativePtr jobArgument, bool toFront, KInt transferMode) {
     Future* future = nullptr;
     Worker* worker = nullptr;
-    {
+    Locker locker(&lock_);
-      Locker locker(&lock_);
 
-      auto it = workers_.find(id);
+    auto it = workers_.find(id);
-      if (it == workers_.end()) return nullptr;
+    if (it == workers_.end()) return nullptr;
-      worker = it->second;
+    worker = it->second;
 
-      future = konanConstructInstance<Future>(nextFutureId());
+    future = konanConstructInstance<Future>(nextFutureId());
-      futures_[future->id()] = future;
+    futures_[future->id()] = future;
-    }
 
     Job job;
-    job.function = reinterpret_cast<KRef (*)(KRef, ObjHeader**)>(jobFunction);
+    if (jobFunction == nullptr) {
-    job.argument = jobArgument;
+      job.kind = JOB_TERMINATE;
-    job.future = future;
+      job.terminationRequest.future = future;
-    job.transferMode = transferMode;
+      job.terminationRequest.waitDelayed = !toFront;
+    } else {
+      job.kind = JOB_REGULAR;
+      job.regularJob.function = reinterpret_cast<KRef (*)(KRef, ObjHeader**)>(jobFunction);
+      job.regularJob.argument = jobArgument;
+      job.regularJob.future = future;
+      job.regularJob.transferMode = transferMode;
+    }
 
     worker->putJob(job, toFront);
 
     return future;
   }
 
+  bool executeJobAfterInWorkerUnlocked(KInt id, KRef operation, KLong afterMicroseconds) {
+    Worker* worker = nullptr;
+    Locker locker(&lock_);
+
+    auto it = workers_.find(id);
+    if (it == workers_.end()) return false;
+    worker = it->second;
+    Job job;
+    job.kind = JOB_EXECUTE_AFTER;
+    job.executeAfter.operation = CreateStablePointer(operation);
+    if (afterMicroseconds == 0) {
+      worker->putJob(job, false);
+    } else {
+      job.executeAfter.whenExecute = konan::getTimeMicros() + afterMicroseconds;
+      worker->putDelayedJob(job);
+    }
+    return true;
+  }
+
   KInt stateOfFutureUnlocked(KInt id) {
     Locker locker(&lock_);
     auto it = futures_.find(id);
@@ -285,7 +416,7 @@ class State {
     struct timeval tv;
     struct timespec ts;
     gettimeofday(&tv, nullptr);
-    KLong nsDelta = millis * 1000000LL;
+    KLong nsDelta = millis * 1000LL * 1000LL;
     ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL;
     ts.tv_sec =  (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL;
     pthread_cond_timedwait(&cond_, &lock_, &ts);
@@ -374,27 +505,46 @@ void* workerRoutine(void* argument) {
     ObjHolder resultHolder;
     while (true) {
       Job job = worker->getJob();
-      if (job.function == nullptr) {
+      if (job.kind == JOB_TERMINATE) {
+        if (job.terminationRequest.waitDelayed) {
+          if (worker->waitDelayed()) {
+            worker->putJob(job, false);
+            continue;
+          }
+        }
         // Termination request, notify the future.
-        job.future->storeResultUnlocked(nullptr, true);
+        job.terminationRequest.future->storeResultUnlocked(nullptr, true);
         theState()->removeWorkerUnlocked(worker->id());
         break;
       }
-      KRef argument = AdoptStablePointer(job.argument, argumentHolder.slot());
+      if (job.kind == JOB_EXECUTE_AFTER) {
+         ObjHolder operationHolder, dummyHolder;
+         KRef obj = DerefStablePointer(job.executeAfter.operation, operationHolder.slot());
+         try {
+           WorkerLaunchpad(obj, dummyHolder.slot());
+         } catch (ExceptionObjHolder& e) {
+           if (worker->errorReporting())
+             ReportUnhandledException(e.obj());
+         }
+         DisposeStablePointer(job.executeAfter.operation);
+         continue;
+      }
+      RuntimeAssert(job.kind == JOB_REGULAR, "Must be regular job");
+      KRef argument = AdoptStablePointer(job.regularJob.argument, argumentHolder.slot());
       KNativePtr result = nullptr;
       bool ok = true;
       try {
-        job.function(argument, resultHolder.slot());
+        job.regularJob.function(argument, resultHolder.slot());
         argumentHolder.clear();
         // Transfer the result.
-        result = transfer(&resultHolder, job.transferMode);
+        result = transfer(&resultHolder, job.regularJob.transferMode);
       } catch (ExceptionObjHolder& e) {
         ok = false;
         if (worker->errorReporting())
           ReportUnhandledException(e.obj());
       }
       // Notify the future.
-      job.future->storeResultUnlocked(result, ok);
+      job.regularJob.future->storeResultUnlocked(result, ok);
     }
   }
 
@@ -417,7 +567,7 @@ KInt currentWorker() {
   return g_currentWorkerId;
 }
 
-KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
+KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
   Job job;
   ObjHolder holder;
   WorkerLaunchpad(producer, holder.slot());
@@ -427,6 +577,11 @@ KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction)
   return future->id();
 }
 
+void executeAfter(KInt id, KRef job, KLong afterMicroseconds) {
+  if (!theState()->executeJobAfterInWorkerUnlocked(id, job, afterMicroseconds))
+    ThrowWorkerInvalidState();
+}
+
 KInt stateOfFuture(KInt id) {
   return theState()->stateOfFutureUnlocked(id);
 }
@@ -476,11 +631,15 @@ KInt stateOfFuture(KInt id) {
   return 0;
 }
 
-KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
+KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
   ThrowWorkerUnsupported();
   return 0;
 }
 
+void executeAfter(KInt id, KRef job, KLong afterMicroseconds) {
+  ThrowWorkerUnsupported();
+}
+
 KInt currentWorker() {
   ThrowWorkerUnsupported();
   return 0;
@@ -531,11 +690,15 @@ KInt Kotlin_Worker_currentInternal() {
 }
 
 KInt Kotlin_Worker_requestTerminationWorkerInternal(KInt id, KBoolean processScheduledJobs) {
-    return requestTermination(id, processScheduledJobs);
+  return requestTermination(id, processScheduledJobs);
 }
 
 KInt Kotlin_Worker_executeInternal(KInt id, KInt transferMode, KRef producer, KNativePtr job) {
-  return schedule(id, transferMode, producer, job);
+  return execute(id, transferMode, producer, job);
+}
+
+void Kotlin_Worker_executeAfterInternal(KInt id, KRef job, KLong afterMicroseconds) {
+  executeAfter(id, job, afterMicroseconds);
 }
 
 KInt Kotlin_Worker_stateOfFuture(KInt id) {

runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt

@@ -42,6 +42,9 @@ external internal fun requestTerminationInternal(id: Int, processScheduledJobs:
 external internal fun executeInternal(
         id: Int, mode: Int, producer: () -> Any?, job: CPointer<CFunction<*>>): Int
 
+@SymbolName("Kotlin_Worker_executeAfterInternal")
+external internal fun executeAfterInternal(id: Int, operation: () -> Unit, afterMicroseconds: Long): Unit
+
 @ExportForCppRuntime
 internal fun ThrowWorkerUnsupported(): Unit =
         throw UnsupportedOperationException("Workers are not supported")

runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt

@@ -61,7 +61,8 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
      * Requests termination of the worker.
      *
      * @param processScheduledJobs controls is we shall wait until all scheduled jobs processed,
-     * or terminate immediately.
+     * or terminate immediately. If there are jobs to be execucted with [executeAfter] their execution
+     * is awaited for.
      */
     public fun requestTermination(processScheduledJobs: Boolean = true) =
             Future<Unit>(requestTerminationInternal(id, processScheduledJobs))
@@ -88,6 +89,20 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
              */
             throw RuntimeException("Shall not be called directly")
 
+
+    /**
+     * Plan job for further execution in the worker. [operation] parameter must be either frozen, or execution to be
+     * planned on the current worker. Otherwise [IllegalStateException] will be thrown.
+     * [afterMicroseconds] defines after how many microseconds delay execution shall happen, 0 means immediately,
+     * on negative values [IllegalArgumentException] is thrown.
+     */
+    public fun executeAfter(afterMicroseconds: Long = 0, operation: () -> Unit): Unit {
+        val current = currentInternal()
+        if (current != id && !operation.isFrozen) throw IllegalStateException("Job for another worker must be frozen")
+        if (afterMicroseconds < 0) throw IllegalArgumentException("Timeout parameter must be non-negative")
+        executeAfterInternal(id, operation, afterMicroseconds)
+    }
+
     override public fun toString(): String = "worker $id"
 
     /**