Workers draft (#655)

runtime/src/main/cpp/Interop.cpp

@@ -73,19 +73,16 @@ void Kotlin_Interop_callWithVarargs(void* codePtr, void* returnValuePtr, FfiType
     ffi_call(&cif, (void (*)())codePtr, returnValuePtr, arguments);
 }
 
-void* Kotlin_Interop_createStablePointer(KRef any) {
-    ::AddRef(any->container());
-    return reinterpret_cast<void*>(any);
+KNativePtr Kotlin_Interop_createStablePointer(KRef any) {
+    return CreateStablePointer(any);
 }
 
-void Kotlin_Interop_disposeStablePointer(void* pointer) {
-    KRef ref = reinterpret_cast<KRef>(pointer);
-    ::Release(ref->container());
+void Kotlin_Interop_disposeStablePointer(KNativePtr pointer) {
+  DisposeStablePointer(pointer);
 }
 
-OBJ_GETTER(Kotlin_Interop_derefStablePointer, void* pointer) {
-    KRef ref = reinterpret_cast<KRef>(pointer);
-    RETURN_OBJ(ref);
+OBJ_GETTER(Kotlin_Interop_derefStablePointer, KNativePtr pointer) {
+  RETURN_RESULT_OF(DerefStablePointer, pointer);
 }
 
 }

runtime/src/main/cpp/Memory.cpp

@@ -224,17 +224,17 @@ ContainerHeaderList collectMutableReferred(ContainerHeader* header) {
   for (int index = 0; index < typeInfo->objOffsetsCount_; index++) {
     ObjHeader** location = reinterpret_cast<ObjHeader**>(
       reinterpret_cast<uintptr_t>(obj + 1) + typeInfo->objOffsets_[index]);
-    ObjHeader* obj = *location;
-    if (obj != nullptr && !isPermanent(obj->container())) {
-      result.push_back(obj->container());
+    ObjHeader* ref = *location;
+    if (ref != nullptr && !isPermanent(ref->container())) {
+      result.push_back(ref->container());
     }
   }
   if (typeInfo == theArrayTypeInfo) {
     ArrayHeader* array = obj->array();
     for (int index = 0; index < array->count_; index++) {
-      ObjHeader* obj = *ArrayAddressOfElementAt(array, index);
-      if (obj != nullptr && !isPermanent(obj->container())) {
-        result.push_back(obj->container());
+      ObjHeader* ref = *ArrayAddressOfElementAt(array, index);
+      if (ref != nullptr && !isPermanent(ref->container())) {
+        result.push_back(ref->container());
       }
     }
   }
@@ -258,6 +258,7 @@ void dumpWorker(const char* prefix, ContainerHeader* header, ContainerHeaderSet*
 void dumpReachable(const char* prefix, const ContainerHeaderSet* roots) {
   ContainerHeaderSet seen;
   for (auto container : *roots) {
+    fprintf(stderr, "%p is root\n", container);
     dumpWorker(prefix, container, &seen);
   }
 }
@@ -590,6 +591,14 @@ void DeinitMemory(MemoryState* memoryState) {
   GarbageCollect();
   delete memoryState->toFree;
   memoryState->toFree = nullptr;
+
+#if OPTIMIZE_GC
+  if (memoryState->toFreeCache != nullptr) {
+    freeMemory(memoryState->toFreeCache);
+    memoryState->toFreeCache = nullptr;
+  }
+#endif
+
 #endif // USE_GC
 
   if (memoryState->allocCount > 0) {
@@ -857,5 +866,55 @@ KInt Kotlin_konan_internal_GC_getThreshold(KRef) {
 #endif
 }
 
+KNativePtr CreateStablePointer(KRef any) {
+  if (any == nullptr) return nullptr;
+  ::AddRef(any->container());
+  return reinterpret_cast<KNativePtr>(any);
+}
+
+void DisposeStablePointer(KNativePtr pointer) {
+  if (pointer == nullptr) return;
+  KRef ref = reinterpret_cast<KRef>(pointer);
+  ::Release(ref->container());
+}
+
+OBJ_GETTER(DerefStablePointer, KNativePtr pointer) {
+  KRef ref = reinterpret_cast<KRef>(pointer);
+  RETURN_OBJ(ref);
+}
+
+OBJ_GETTER(AdoptStablePointer, KNativePtr pointer) {
+  __sync_synchronize();
+  KRef ref = reinterpret_cast<KRef>(pointer);
+  // Somewhat hacky.
+  *OBJ_RESULT = ref;
+  return ref;
+}
+
+bool ClearSubgraphReferences(ObjHeader* root, bool checked) {
+#if USE_GC
+  if (root != nullptr) {
+    auto state = memoryState;
+    auto container = root->container();
+    ContainerHeaderList todo;
+    ContainerHeaderSet subgraph;
+    todo.push_back(container);
+    while (todo.size() > 0) {
+      auto header = todo.back();
+      todo.pop_back();
+      if (subgraph.count(header) != 0)
+        continue;
+      subgraph.insert(header);
+      removeFreeable(state, header);
+      auto children = collectMutableReferred(header);
+      for (auto child : children) {
+        todo.push_back(child);
+      }
+    }
+  }
+#endif  // USE_GC
+  // TODO: perform trial deletion starting from this root, if in checked mode.
+  return true;
+}
 
 } // extern "C"

runtime/src/main/cpp/Memory.h

@@ -353,6 +353,18 @@ void ReleaseRefs(ObjHeader** start, int count) RUNTIME_NOTHROW;
 void LeaveFrame(ObjHeader** start, int count) RUNTIME_NOTHROW;
 // Collect garbage, which cannot be found by reference counting (cycles).
 void GarbageCollect() RUNTIME_NOTHROW;
+// Clears object subgraph references from memory subsystem, and optionally
+// checks if subgraph referenced by given root is disjoint from the rest of
+// object graph, i.e. no external references exists.
+bool ClearSubgraphReferences(ObjHeader* root, bool checked) RUNTIME_NOTHROW;
+// Creates stable pointer out of the object.
+void* CreateStablePointer(ObjHeader* obj) RUNTIME_NOTHROW;
+// Disposes stable pointer to the object.
+void DisposeStablePointer(void* pointer) RUNTIME_NOTHROW;;
+// Translate stable pointer to object reference.
+OBJ_GETTER(DerefStablePointer, void*) RUNTIME_NOTHROW;
+// Move stable pointer ownership.
+OBJ_GETTER(AdoptStablePointer, void*) RUNTIME_NOTHROW;
 
 #ifdef __cplusplus
 }

runtime/src/main/cpp/Types.h

@@ -30,6 +30,7 @@ typedef int32_t KInt;
 typedef int64_t KLong;
 typedef float   KFloat;
 typedef double  KDouble;
+typedef void*   KNativePtr;
 
 typedef ObjHeader* KRef;
 typedef const ObjHeader* KConstRef;

runtime/src/main/cpp/Worker.cpp

@@ -0,0 +1,523 @@
+/*
+ * Copyright 2010-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#define WITH_WORKERS 1
+
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+
+#if WITH_WORKERS
+#include <pthread.h>
+#include <sys/time.h>
+
+#include <deque>
+#include <unordered_map>
+#endif
+
+#include "Assert.h"
+#include "Memory.h"
+#include "Runtime.h"
+#include "Types.h"
+
+extern "C" {
+
+void ThrowWorkerInvalidState();
+void ThrowWorkerUnsupported();
+OBJ_GETTER(WorkerLaunchpad, KRef);
+
+}  // extern "C"
+
+namespace {
+
+#if WITH_WORKERS
+
+enum {
+  INVALID = 0,
+  SCHEDULED = 1,
+  COMPUTED = 2,
+  CANCELLED = 3
+};
+
+enum {
+  CHECKED = 0,
+  UNCHECKED = 1
+};
+
+KNativePtr transfer(KRef object, KInt mode) {
+  switch (mode) {
+    case CHECKED:
+    case UNCHECKED:
+      if (!ClearSubgraphReferences(object, mode == CHECKED)) {
+        ThrowWorkerInvalidState();
+        return nullptr;
+      }
+      return object;
+  }
+  return nullptr;
+}
+
+class Locker {
+ public:
+  explicit Locker(pthread_mutex_t* lock) : lock_(lock) {
+    pthread_mutex_lock(lock_);
+  }
+  ~Locker() {
+     pthread_mutex_unlock(lock_);
+  }
+
+ private:
+  pthread_mutex_t* lock_;
+};
+
+class Future {
+ public:
+  Future(KInt id) : state_(SCHEDULED), id_(id) {
+    pthread_mutex_init(&lock_, nullptr);
+    pthread_cond_init(&cond_, nullptr);
+  }
+
+  ~Future() {
+    pthread_mutex_destroy(&lock_);
+    pthread_cond_destroy(&cond_);
+  }
+
+  OBJ_GETTER0(consumeResultUnlocked) {
+    Locker locker(&lock_);
+    while (state_ == SCHEDULED) {
+      pthread_cond_wait(&cond_, &lock_);
+    }
+    auto result = AdoptStablePointer(result_, OBJ_RESULT);
+    result_ = nullptr;
+    return result;
+  }
+
+  void storeResultUnlocked(KNativePtr result);
+
+  void cancelUnlocked();
+
+  // Those are called with the lock taken.
+  KInt state() const { return state_; }
+  KInt id() const { return id_; }
+
+ private:
+  // State of future execution.
+  KInt state_;
+  // Integer id of the future.
+  KInt id_;
+  // Stable pointer with future's result.
+  KNativePtr result_;
+  // Lock and condition for waiting on the future.
+  pthread_mutex_t lock_;
+  pthread_cond_t cond_;
+};
+
+struct Job {
+  KRef (*function)(KRef, ObjHeader**);
+  KNativePtr argument;
+  Future* future;
+  KInt transferMode;
+};
+
+class Worker {
+ public:
+  Worker(KInt id) : id_(id) {
+    pthread_mutex_init(&lock_, nullptr);
+    pthread_cond_init(&cond_, nullptr);
+  }
+
+  ~Worker() {
+    // Cleanup jobs in queue.
+    for (auto job : queue_) {
+      DisposeStablePointer(job.argument);
+      job.future->cancelUnlocked();
+    }
+
+    pthread_mutex_destroy(&lock_);
+    pthread_cond_destroy(&cond_);
+  }
+
+  void putJob(Job job, bool toFront) {
+    Locker locker(&lock_);
+    if (toFront)
+      queue_.push_front(job);
+    else
+      queue_.push_back(job);
+    pthread_cond_signal(&cond_);
+  }
+
+  Job getJob() {
+    Locker locker(&lock_);
+    while (queue_.size() == 0) {
+      pthread_cond_wait(&cond_, &lock_);
+    }
+    auto result = queue_.front();
+    queue_.pop_front();
+    return result;
+  }
+
+  KInt id() const { return id_; }
+
+ private:
+  KInt id_;
+  std::deque<Job> queue_;
+  // Lock and condition for waiting on the queue.
+  pthread_mutex_t lock_;
+  pthread_cond_t cond_;
+};
+
+class State {
+ public:
+  State() {
+    pthread_mutex_init(&lock_, nullptr);
+    pthread_cond_init(&cond_, nullptr);
+
+    currentWorkerId_ = 1;
+    currentFutureId_ = 1;
+    currentVersion_ = 0;
+  }
+
+  ~State() {
+    // TODO: some sanity check here?
+    pthread_mutex_destroy(&lock_);
+    pthread_cond_destroy(&cond_);
+  }
+
+  Worker* addWorkerUnlocked() {
+    Locker locker(&lock_);
+    Worker* worker = new Worker(nextWorkerId());
+    if (worker == nullptr) return nullptr;
+    workers_[worker->id()] = worker;
+    return worker;
+  }
+
+  void removeWorkerUnlocked(KInt id) {
+    Locker locker(&lock_);
+    auto it = workers_.find(id);
+    if (it == workers_.end()) return;
+    workers_.erase(it);
+  }
+
+  Future* addJobToWorkerUnlocked(
+      KInt id, KNativePtr jobFunction, KNativePtr jobArgument, bool toFront, KInt transferMode) {
+    Future* future = nullptr;
+    Worker* worker = nullptr;
+    {
+      Locker locker(&lock_);
+
+      auto it = workers_.find(id);
+      if (it == workers_.end()) return nullptr;
+      worker = it->second;
+
+      future = new Future(nextFutureId());
+      futures_[future->id()] = future;
+    }
+
+    Job job;
+    job.function = reinterpret_cast<KRef (*)(KRef, ObjHeader**)>(jobFunction);
+    job.argument = jobArgument;
+    job.future = future;
+    job.transferMode = transferMode;
+
+    worker->putJob(job, toFront);
+
+    return future;
+  }
+
+  KInt stateOfFutureUnlocked(KInt id) {
+    Locker locker(&lock_);
+    auto it = futures_.find(id);
+    if (it == futures_.end()) return INVALID;
+    return it->second->state();
+  }
+
+  OBJ_GETTER(consumeFutureUnlocked, KInt id) {
+    Future* future = nullptr;
+    {
+      Locker locker(&lock_);
+      auto it = futures_.find(id);
+      if (it == futures_.end()) ThrowWorkerInvalidState();
+      future = it->second;
+    }
+
+    KRef result = future->consumeResultUnlocked(OBJ_RESULT);
+
+    {
+       Locker locker(&lock_);
+       auto it = futures_.find(id);
+       if (it != futures_.end()) {
+         futures_.erase(it);
+         delete future;
+       }
+    }
+
+    return result;
+  }
+
+  KBoolean waitForAnyFuture(KInt version, KInt millis) {
+    Locker locker(&lock_);
+    if (version != currentVersion_) return false;
+
+    if (millis < 0) {
+      pthread_cond_wait(&cond_, &lock_);
+      return true;
+    }
+    struct timeval tv;
+    struct timespec ts;
+    gettimeofday(&tv, nullptr);
+    KLong nsDelta = millis * 1000000LL;
+    ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL;
+    ts.tv_sec =  (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL;
+    pthread_cond_timedwait(&cond_, &lock_, &ts);
+    return true;
+  }
+
+  void signalAnyFuture() {
+    {
+      Locker locker(&lock_);
+      currentVersion_++;
+    }
+    pthread_cond_broadcast(&cond_);
+  }
+
+  KInt versionToken() {
+    Locker locker(&lock_);
+    return currentVersion_;
+  }
+
+  // All those called with lock taken.
+  KInt nextWorkerId() { return currentWorkerId_++; }
+  KInt nextFutureId() { return currentFutureId_++; }
+
+ private:
+  pthread_mutex_t lock_;
+  pthread_cond_t cond_;
+  std::unordered_map<KInt, Future*> futures_;
+  std::unordered_map<KInt, Worker*> workers_;
+  KInt currentWorkerId_;
+  KInt currentFutureId_;
+  KInt currentVersion_;
+};
+
+State* theState() {
+  static State* state = nullptr;
+
+  if (state != nullptr) {
+    return state;
+  }
+
+  State* result = new State();
+
+  State* old = __sync_val_compare_and_swap(&state, nullptr, result);
+  if (old != nullptr) {
+    delete result;
+    // Someone else inited this data.
+    return old;
+  }
+  return state;
+}
+
+void Future::storeResultUnlocked(KNativePtr result) {
+  {
+    Locker locker(&lock_);
+    state_ = COMPUTED;
+    result_ = result;
+  }
+  pthread_cond_signal(&cond_);
+  theState()->signalAnyFuture();
+}
+
+void Future::cancelUnlocked() {
+  {
+    Locker locker(&lock_);
+    state_ = CANCELLED;
+    result_ = nullptr;
+  }
+  pthread_cond_signal(&cond_);
+  theState()->signalAnyFuture();
+}
+
+
+void* workerRoutine(void* argument) {
+  Worker* worker = reinterpret_cast<Worker*>(argument);
+
+  RuntimeState* state = InitRuntime();
+  while (true) {
+    Job job = worker->getJob();
+    if (job.function == nullptr) {
+       // Termination request, notify the future.
+      job.future->storeResultUnlocked(nullptr);
+      theState()->removeWorkerUnlocked(worker->id());
+      break;
+    }
+    ObjHolder argumentHolder;
+    KRef argument = AdoptStablePointer(job.argument, argumentHolder.slot());
+    // Note that this is a bit hacky, as we must not auto-release resultRef,
+    // so we don't use ObjHolder.
+    // It is so, as ownership is transferred.
+    KRef resultRef = nullptr;
+    job.function(argument, &resultRef);
+    // Transfer the result.
+    KNativePtr result = transfer(resultRef, job.transferMode);
+    // Notify the future.
+    job.future->storeResultUnlocked(result);
+  }
+
+  DeinitRuntime(state);
+
+  delete worker;
+
+  return nullptr;
+}
+
+KInt startWorker() {
+  Worker* worker = theState()->addWorkerUnlocked();
+  if (worker == nullptr) return -1;
+  pthread_t thread = 0;
+  pthread_create(&thread, nullptr, workerRoutine, worker);
+  return worker->id();
+}
+
+KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
+  Job job;
+  // Note that this is a bit hacky, as we must not auto-release jobArgumentRef,
+  // so we don't use ObjHolder.
+  KRef jobArgumentRef = nullptr;
+  WorkerLaunchpad(producer, &jobArgumentRef);
+  KNativePtr jobArgument = transfer(jobArgumentRef, transferMode);
+  Future* future = theState()->addJobToWorkerUnlocked(id, jobFunction, jobArgument, false, transferMode);
+  if (future == nullptr) ThrowWorkerInvalidState();
+  return future->id();
+}
+
+OBJ_GETTER(shallowCopy, KConstRef object) {
+  if (object == nullptr) RETURN_OBJ(nullptr);
+
+  const TypeInfo* typeInfo = object->type_info();
+  bool isArray = typeInfo->instanceSize_ < 0;
+  KRef result = isArray ?
+      AllocArrayInstance(typeInfo, object->array()->count_, OBJ_RESULT) :
+      AllocInstance(typeInfo, OBJ_RESULT);
+  // TODO: what to do when object references exist.
+  if (isArray) {
+    RuntimeAssert(object->array()->count_ == 0 || typeInfo != theArrayTypeInfo, "Object array copy unimplemented");
+    memcpy(result->array() + 1, object->array() + 1, ArrayDataSizeBytes(object->array()));
+  } else {
+    RuntimeAssert(typeInfo->objOffsetsCount_ == 0, "Object reference copy unimplemented");
+    memcpy(result + 1, object + 1, typeInfo->instanceSize_);
+  }
+  return result;
+}
+
+KInt stateOfFuture(KInt id) {
+  return theState()->stateOfFutureUnlocked(id);
+}
+
+OBJ_GETTER(consumeFuture, KInt id) {
+  RETURN_RESULT_OF(theState()->consumeFutureUnlocked, id);
+}
+
+KInt requestTermination(KInt id, KBoolean processScheduledJobs) {
+  Future* future = theState()->addJobToWorkerUnlocked(
+      id, nullptr, nullptr, /* toFront = */ !processScheduledJobs, UNCHECKED);
+  if (future == nullptr) ThrowWorkerInvalidState();
+  return future->id();
+}
+
+KBoolean waitForAnyFuture(KInt version, KInt millis) {
+  return theState()->waitForAnyFuture(version, millis);
+}
+
+KInt versionToken() {
+  return theState()->versionToken();
+}
+
+#else
+
+KInt startWorker() {
+  ThrowWorkerUnsupported();
+  return -1;
+}
+
+OBJ_GETTER(shallowCopy, KConstRef object) {
+  ThrowWorkerUnsupported();
+  RETURN_OBJ(nullptr);
+}
+
+KInt stateOfFuture(KInt id) {
+  ThrowWorkerUnsupported();
+  return 0;
+}
+
+OBJ_GETTER(consumeFuture, KInt id) {
+  ThrowWorkerUnsupported();
+  RETURN_OBJ(nullptr);
+}
+
+KInt requestTermination(KInt id, KBoolean processScheduledJobs) {
+  ThrowWorkerUnsupported();
+  return -1;
+}
+
+KBoolean waitForAnyFuture(KInt versionToken, KInt millis) {
+  ThrowWorkerUnsupported();
+  return false;
+}
+
+KInt versionToken() {
+  ThrowWorkerUnsupported();
+  return 0;
+}
+
+#endif  // WITH_WORKERS
+
+}  // namespace
+
+extern "C" {
+
+KInt Kotlin_Worker_startInternal() {
+  return startWorker();
+}
+
+KInt Kotlin_Worker_requestTerminationWorkerInternal(KInt id, KBoolean processScheduledJobs) {
+    return requestTermination(id, processScheduledJobs);
+}
+
+KInt Kotlin_Worker_scheduleInternal(KInt id, KInt transferMode, KRef producer, KNativePtr job) {
+  return schedule(id, transferMode, producer, job);
+}
+
+OBJ_GETTER(Kotlin_Worker_shallowCopyInternal, KConstRef object) {
+  RETURN_RESULT_OF(shallowCopy, object);
+}
+
+KInt Kotlin_Worker_stateOfFuture(KInt id) {
+  return stateOfFuture(id);
+}
+
+OBJ_GETTER(Kotlin_Worker_consumeFuture, KInt id) {
+  RETURN_RESULT_OF(consumeFuture, id);
+}
+
+KBoolean Kotlin_Worker_waitForAnyFuture(KInt versionToken, KInt millis) {
+  return waitForAnyFuture(versionToken, millis);
+}
+
+KInt Kotlin_Worker_versionToken() {
+  return versionToken();
+}
+
+
+}  // extern "C"

runtime/src/main/kotlin/konan/Annotations.kt

@@ -33,13 +33,16 @@ annotation class SymbolName(val name: String)
 //@Retention(AnnotationRetention.SOURCE)
 annotation class ExportTypeInfo(val name: String)
 
+/**
+ * * If lambda shall be carefully lowered by the compiler.
+ */
+annotation class VolatileLambda
 
 /**
  * Preserve the function entry point during global optimizations
  */
 public annotation class Used
 
-
 /**
  * Need to be fixed because of reification support.
  */

runtime/src/main/kotlin/konan/worker/Worker.kt

@@ -0,0 +1,243 @@
+/*
+ * Copyright 2010-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package konan.worker
+
+import konan.SymbolName
+import konan.internal.ExportForCppRuntime
+import kotlinx.cinterop.*
+
+/**
+ *      Workers: theory of operations.
+ *
+ *  Worker represent asynchronous and concurrent computation, usually performed by other threads
+ * in the same process. Object passing between workers is performed using transfer operation, so that
+ * object graph belongs to one worker at the time, but can be disconnected and reconnected as needed.
+ * See 'Object Transfer Basics' below for more details on how objects shall be transferred.
+ * This approach ensures that no concurrent access happens to same object, while data may flow between
+ * workers as needed.
+ */
+
+/**
+ * State of the future object.
+ */
+enum class FutureState(val value: Int) {
+    INVALID(0),
+    // Future is scheduled for execution.
+    SCHEDULED(1),
+    // Future result is computed.
+    COMPUTED(2),
+    // Future is cancelled.
+    CANCELLED(3)
+}
+
+/**
+ *  Object Transfer Basics.
+ *
+ *  Objects can be passed between threads in one of two possible modes.
+ *
+ *    - CHECKED - object subgraph is checked to be not reachable by other globals or locals, and passed
+ *      if so, otherwise an exception is thrown
+ *    - UNCHECKED - object is blindly passed to another worker, if there are references
+ *      left in the passing worker - it may lead to crash or program malfunction
+ *
+ *   Checked mode checks if object is no longer used in passing worker, using memory-management
+ *  specific algorithm (ARC implementation relies on trial deletion on object graph rooted in
+ *  passed object), and throws IllegalStateException if object graph rooted in transferred object
+ *  is reachable by some other means,
+ *
+ *   Unchecked mode, intended for most performance crititcal operations, where object graph ownership
+ *  is expected to be correct (such as application debugged earlier in CHECKED mode), just transfers
+ *  ownership without further checks.
+ *
+ *   Note, that for some cases cycle collection need to be done to ensure that dead cycles do not affect
+ *  reachability of passed object graph. See `konan.internal.GC.collect()`.
+ *
+ */
+enum class TransferMode(val value: Int) {
+    CHECKED(0),
+    UNCHECKED(1) // USE UNCHECKED MODE ONLY IF ABSOLUTELY SURE WHAT YOU'RE DOING!!!
+}
+
+/**
+ * Unique identifier of the worker. Workers can be used from other workers.
+ */
+typealias WorkerId = Int
+/**
+ * Unique identifier of the future. Futures can be used from other workers.
+ */
+typealias FutureId = Int
+
+/**
+ * Class representing abstract computation, whose result may become available in the future.
+ */
+// TODO: make me value class!
+class Future<T> internal constructor(val id: FutureId) {
+    /**
+     * Blocks execution until the future is ready.
+     */
+    fun consume(code: (T) -> Unit) {
+        when (state) {
+            FutureState.SCHEDULED, FutureState.COMPUTED -> {
+                val value = consumeFuture(id) as T
+                code(value)
+            }
+            FutureState.INVALID ->
+                throw IllegalStateException("Future is in an invalid state: $state")
+            FutureState.CANCELLED ->
+                throw IllegalStateException("Future is cancelled")
+        }
+    }
+
+    val state: FutureState
+            get() = FutureState.values()[stateOfFuture(id)]
+
+    override fun equals(other: Any?): Boolean {
+        return (other is Future<*>) && (id == other.id)
+    }
+
+    override fun hashCode(): Int {
+        return id
+    }
+}
+
+/**
+ * Class representing worker.
+ */
+// TODO: make me value class!
+class Worker(val id: WorkerId) {
+    /**
+     * Requests termination of the worker. `processScheduledJobs` controls is we shall wait
+     * until all scheduled jobs processed, or terminate immediately.
+     */
+    fun requestTermination(processScheduledJobs: Boolean = true) =
+            Future<Any?>(requestTerminationInternal(id, processScheduledJobs))
+
+    /**
+     * Schedule a job for further execution in the worker. Schedule is a two-phase operation,
+     * first `producer` function is executed, and resulting object and whatever it refers to
+     * is analyzed for being an isolated object subgraph, if in checked mode.
+     * Afterwards, this disconnected object graph and `job` function pointer is being added to jobs queue
+     * of the selected worker. Note that `job` must not capture any state itself, so that whole state is
+     * explicitly stored in object produced by `producer`. Scheduled job is being executed by the worker,
+     * and result of such a execution is being disconnected from worker's object graph. Whoever will consume
+     * the future, can use result of worker's computations.
+     */
+    fun <T1, T2> schedule(mode: TransferMode, producer: () -> T1,
+                          @VolatileLambda job: (T1) -> T2): Future<T2> =
+            /**
+             * This function is a magical operation, handled by lowering in the compiler, and replaced with call to
+             *   scheduleImpl(worker, mode, producer, job)
+             * but first ensuring that `job` parameter  doesn't capture any state.
+             */
+            throw RuntimeException("Shall not be called directly")
+
+    override fun equals(other: Any?): Boolean {
+        return (other is Worker) && (id == other.id)
+    }
+
+    override fun hashCode(): Int {
+        return id
+    }
+}
+
+/**
+ * Start new scheduling primitive, such as thread, to accept new tasks via `schedule` interface.
+ * Typically new worker may be needed for computations offload to another core, for IO it may be
+ * better to use non-blocking IO combined with more lightweight coroutines.
+ */
+fun startWorker() : Worker = Worker(startInternal())
+
+/**
+ * Wait for availability of futures in the collection. Returns set with all futures which have
+ * value available for the consumption.
+ */
+fun <T> Collection<Future<T>>.waitForMultipleFutures(millis: Int) : Set<Future<T>> {
+    val result = mutableSetOf<Future<T>>()
+
+    while (true) {
+        val versionToken = versionToken()
+        for (future in this) {
+            if (future.state == FutureState.COMPUTED) {
+                result += future
+            }
+        }
+        if (result.isNotEmpty()) return result
+
+        if (waitForAnyFuture(versionToken, millis)) break
+    }
+
+    for (future in this) {
+        if (future.state == FutureState.COMPUTED) {
+            result += future
+        }
+    }
+
+    return result
+}
+
+/**
+ * Creates verbatim *shallow* copy of passed object, use carefully to create disjoint object graph.
+ */
+fun <T> T.shallowCopy(): T = shallowCopyInternal(this) as T
+
+/**
+ * Creates verbatim *deep* copy of passed object's graph, use *VERY* carefully to create disjoint object graph.
+ * Note that this function could potentially duplicate a lot of objects.
+ */
+fun <T> T.deepCopy(): T = TODO()
+
+// Implementation details.
+@konan.internal.ExportForCompiler
+internal fun scheduleImpl(worker: Worker, mode: TransferMode, producer: () -> Any?,
+                          job: CPointer<CFunction<*>>) : Future<Any?> =
+        Future<Any?>(scheduleInternal(worker.id, mode.value, producer, job))
+
+@SymbolName("Kotlin_Worker_startInternal")
+external internal fun startInternal() : WorkerId
+
+@SymbolName("Kotlin_Worker_requestTerminationWorkerInternal")
+external internal fun requestTerminationInternal(id: WorkerId, processScheduledJobs: Boolean): FutureId
+
+@SymbolName("Kotlin_Worker_scheduleInternal")
+external internal fun scheduleInternal(
+        id: WorkerId, mode: Int, producer: () -> Any?, job: CPointer<CFunction<*>>) : FutureId
+
+@SymbolName("Kotlin_Worker_shallowCopyInternal")
+external internal fun shallowCopyInternal(value: Any?) : Any?
+
+@SymbolName("Kotlin_Worker_stateOfFuture")
+external internal fun stateOfFuture(id: FutureId): Int
+
+@SymbolName("Kotlin_Worker_consumeFuture")
+external internal fun consumeFuture(id: FutureId): Any?
+
+@SymbolName("Kotlin_Worker_waitForAnyFuture")
+external internal fun waitForAnyFuture(versionToken: Int, millis: Int): Boolean
+
+@SymbolName("Kotlin_Worker_versionToken")
+external internal fun versionToken(): Int
+
+@ExportForCppRuntime
+internal fun ThrowWorkerUnsupported(): Unit =
+        throw UnsupportedOperationException("Workers are not supported")
+
+@ExportForCppRuntime
+internal fun ThrowWorkerInvalidState(): Unit =
+        throw IllegalStateException("Illegal transfer state")
+
+@ExportForCppRuntime
+internal fun WorkerLaunchpad(function: () -> Any?) = function()