azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

samples: Remove concurrent samples

Browse Source
This commit is contained in:
Ilya Matveev
2017-10-23 12:08:12 +03:00
committed by ilmat192
parent 2dde9eda55
commit cd351fda44
9 changed files with 0 additions and 579 deletions
Download Patch File Download Diff File Expand all files Collapse all files
samples/concurrent/README.md
-15
View File
@@ -1,15 +0,0 @@
# Concurrent
This example shows how to implement concurrent programming in Kotlin/Native.
In this example we start multiple threads running concurrently and exchange messages with them.
To build cpp use `./buildCpp.sh`.
To build kotlin native use `../gradlew build` or `./build.sh`.
To run use `../gradlew run`
Alternatively you can run artifact directly
./build/konan/bin/MessageChannel/MessageChannel.kexe
It will print all passed messages.
samples/concurrent/build.gradle
-28
View File
@@ -1,28 +0,0 @@
apply plugin: 'konan'
konanInterop {
MessageChannel {
includeDirs "${project.projectDir}/src/main/cpp"
}
}
konanArtifacts {
MessageChannel {
useInterop "MessageChannel"
nativeLibrary "${project.buildDir.canonicalPath}/clang/MessageChannel.bc"
}
}
task compileCpp(type: Exec) {
dependsOn 'downloadKonanCompiler'
workingDir project.getProjectDir()
commandLine './buildCpp.sh'
}
compileKonanMessageChannel {
dependsOn 'compileCpp'
}
downloadKonanCompiler {
downloadDependencies = true
}
samples/concurrent/build.sh
-30
View File
@@ -1,30 +0,0 @@
#!/usr/bin/env bash
DIR=$(cd "$(dirname "${BASH_SOURCE[0]}" )" && pwd )
PATH=$DIR/../../dist/bin:$DIR/../../bin:$PATH
if [ x$TARGET == x ]; then
case "$OSTYPE" in
darwin*) TARGET=macbook ;;
linux*) TARGET=linux ;;
*) echo "unknown: $OSTYPE" && exit 1;;
esac
fi
var=CFLAGS_${TARGET}
CFLAGS=${!var}
var=COMPILER_ARGS_${TARGET}
COMPILER_ARGS=${!var} # add -opt for an optimized build.
mkdir -p $DIR/build/c_interop
mkdir -p $DIR/build/bin
$DIR/buildCpp.sh
cinterop -def $DIR/src/main/c_interop/MessageChannel.def -copt "-I$DIR/src/main/cpp" -target $TARGET \
-o $DIR/build/c_interop/MessageChannel || exit 1
konanc $DIR/src/main/kotlin/Concurrent.kt -target $TARGET -library $DIR/build/c_interop/MessageChannel \
-nativelibrary $DIR/build/clang/MessageChannel.bc -o $DIR/build/bin/Concurrent || exit 1
echo "Artifact path is $DIR/build/bin/Concurrent.kexe"
samples/concurrent/buildCpp.sh
-26
View File
@@ -1,26 +0,0 @@
#!/usr/bin/env bash
DIR=$(cd "$(dirname "${BASH_SOURCE[0]}" )" && pwd )
PATH=$DIR/../../dist/bin:$DIR/../../bin:$PATH
DEPS=$(dirname `type -p konanc`)/../dependencies
if [ x$TARGET == x ]; then
case "$OSTYPE" in
darwin*) TARGET=macbook ;;
linux*) TARGET=linux ;;
*) echo "unknown: $OSTYPE" && exit 1;;
esac
fi
# Ensure that C compiler for the target is available.
konanc --check_dependencies -target $TARGET || exit 1
CLANG_linux=$DEPS/clang-llvm-3.9.0-linux-x86-64/bin/clang++
CLANG_macbook=$DEPS/clang-llvm-3.9.0-darwin-macos/bin/clang++
var=CLANG_${TARGET}
CLANG=${!var}
mkdir -p $DIR/build/clang/
$CLANG -H -std=c++11 -c $DIR/src/main/cpp/MessageChannel.cpp -o $DIR/build/clang/MessageChannel.bc -emit-llvm || exit 1
samples/concurrent/src/main/c_interop/MessageChannel.def
-2
View File
@@ -1,2 +0,0 @@
headers = MessageChannel.h
headerFilter = MessageChannel.h
samples/concurrent/src/main/cpp/MessageChannel.cpp
-306
View File
@@ -1,306 +0,0 @@
/*
* 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.
*/
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <cassert>
#include <deque>
#include <unordered_map>
#include <unordered_set>
#include "MessageChannel.h"
namespace {
// Konan entry point.
extern "C" int Konan_main(int argc, const char** argv);
struct WorkerState {
WorkerState(worker_id_t id, int argc, const char** argv)
: id_(id), argc_(argc + 1), argv_(nullptr) {
printf("create with %d\n", id);
argv_ = reinterpret_cast<char**>(malloc(sizeof(char*) * argc_));
argv_[0] = strdup("worker");
for (auto i = 0; i < argc; i++) {
argv_[i + 1] = strdup(argv[i]);
}
pthread_mutex_init(&mutex_, nullptr);
pthread_cond_init(&cond_, nullptr);
blocked_on_message_ = nullptr;
blocked_message_used_ = false;
}
~WorkerState() {
for (auto i = 0; i < argc_; i++) {
free(argv_[i]);
}
if (argv_) free(argv_);
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&cond_);
}
int GetMessageNotEmptyLocked(Message* message);
bool HasMessageLocked() const {
return (blocked_on_message_ != nullptr && blocked_message_used_) ||
!queue_.empty();
}
std::deque<Message*> queue_;
Message* blocked_on_message_;
bool blocked_message_used_;
pthread_mutex_t mutex_;
pthread_cond_t cond_;
worker_id_t id_;
pthread_t tid_;
int argc_;
char** argv_;
};
__thread WorkerState* current_worker = nullptr;
struct WorkersState {
WorkersState() : current_worker_id_(1) {
pthread_mutex_init(&mutex_, nullptr);
auto thisWorker = new WorkerState(current_worker_id_++, 0, nullptr);
AddWorker(pthread_self(), thisWorker);
current_worker = thisWorker;
}
~WorkersState() {
for (auto worker : workers_) delete worker;
pthread_mutex_destroy(&mutex_);
}
void AddWorkerLocked(pthread_t tid, WorkerState* state) {
workers_.insert(state);
state->tid_ = tid;
id_map_[state->id_] = state;
thread_map_[state->tid_] = state;
}
void AddWorker(pthread_t tid, WorkerState* state) {
pthread_mutex_lock(&mutex_);
AddWorkerLocked(tid, state);
pthread_mutex_unlock(&mutex_);
}
void RemoveWorkerLocked(WorkerState* worker) {
workers_.erase(worker);
id_map_.erase(worker->id_);
thread_map_.erase(worker->tid_);
delete worker;
}
void RemoveWorker(WorkerState* worker) {
pthread_mutex_lock(&mutex_);
RemoveWorkerLocked(worker);
pthread_mutex_unlock(&mutex_);
}
pthread_mutex_t mutex_;
worker_id_t current_worker_id_;
std::unordered_map<worker_id_t, WorkerState*> id_map_;
std::unordered_map<pthread_t, WorkerState*> thread_map_;
std::unordered_set<WorkerState*> workers_;
};
WorkersState* getWorkers() {
static WorkersState* instance = new WorkersState();
return instance;
}
void copyMessageSteal(Message* destination, Message* source) {
destination->kind_ = source->kind_;
if (source->data_size_ > 0) {
free(destination->data_);
destination->data_ = source->data_;
destination->data_capacity_ = source->data_capacity_;
source->data_ = nullptr;
}
destination->data_size_ = source->data_size_;
}
void copyMessageNoSteal(Message* destination, const Message* source) {
destination->kind_ = source->kind_;
if (destination->data_capacity_ < source->data_size_) {
if (destination->data_) free(destination->data_);
destination->data_ = malloc(source->data_size_);
destination->data_capacity_ = source->data_size_;
}
memcpy(destination->data_, source->data_, source->data_size_);
destination->data_size_ = source->data_size_;
}
void* ThreadRunner(void* arg) {
WorkerState* state = reinterpret_cast<WorkerState*>(arg);
current_worker = state;
Konan_main(state->argc_, const_cast<const char**>(state->argv_));
getWorkers()->RemoveWorker(current_worker);
return nullptr;
}
void copyOrEnqueueMessageLocked(WorkerState* sender, WorkerState* receiver, const Message* message) {
bool use_blocked = receiver->blocked_on_message_ != nullptr &&
!receiver->blocked_message_used_;
Message* result =
use_blocked ? receiver->blocked_on_message_ : CreateMessage(message->data_size_);
result->source_ = sender->id_;
result->destination_ = receiver->id_;
copyMessageNoSteal(result, message);
if (use_blocked)
receiver->blocked_message_used_ = true;
else
receiver->queue_.push_back(result);
}
int WorkerState::GetMessageNotEmptyLocked(Message* result) {
bool blocked_used = blocked_on_message_ != nullptr && blocked_message_used_;
if (blocked_used) {
assert(result == blocked_on_message_);
blocked_on_message_ = nullptr;
blocked_message_used_ = false;
return 0;
}
Message* from_queue = queue_.front();
result->source_ = from_queue->source_;
result->destination_ = from_queue->destination_;
copyMessageSteal(result, from_queue);
queue_.pop_front();
ReleaseMessage(from_queue);
return 0;
}
} // namespace
#ifdef __cplusplus
extern "C" {
#endif
Message* CreateMessage(int64_t data_capacity) {
auto result = reinterpret_cast<Message*>(calloc(1, sizeof(Message)));
result->data_size_ = 0;
if (data_capacity == 0) {
result->data_ = nullptr;
} else {
result->data_ = malloc(data_capacity);
}
result->data_capacity_ = data_capacity;
return result;
}
void ReleaseMessage(Message* message) {
if (message->data_ != nullptr) free(message->data_);
free(message);
}
worker_id_t CreateWorker(const char* where, int argc, const char** argv) {
auto workers = getWorkers();
worker_id_t id = INVALID_WORKER;
pthread_mutex_lock(&workers->mutex_);
WorkerState* state = new WorkerState(workers->current_worker_id_++, argc, argv);
pthread_t tid;
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
int rc = pthread_create(&tid, &attr, ThreadRunner, state);
if (rc != 0) {
printf("pthread_create(): %d\n", rc);
delete state;
} else {
workers->AddWorkerLocked(tid, state);
id = state->id_;
}
pthread_mutex_unlock(&workers->mutex_);
return id;
}
worker_id_t CurrentWorker() {
return current_worker->id_;
}
int SendMessage(worker_id_t destination_id, const Message* message) {
auto workers = getWorkers();
int rc = 0;
pthread_mutex_lock(&workers->mutex_);
auto destination = workers->id_map_[destination_id];
if (destination != nullptr) {
pthread_mutex_lock(&destination->mutex_);
copyOrEnqueueMessageLocked(current_worker, destination, message);
pthread_cond_signal(&destination->cond_);
pthread_mutex_unlock(&destination->mutex_);
} else {
rc = -1;
}
pthread_mutex_unlock(&workers->mutex_);
return rc;
}
int GetMessage(Message* message, int timeout_ms) {
auto worker = current_worker;
int result = -1;
pthread_mutex_lock(&worker->mutex_);
if (!worker->queue_.empty()) {
result = worker->GetMessageNotEmptyLocked(message);
} else {
worker->blocked_on_message_ = message;
worker->blocked_message_used_ = false;
if (timeout_ms < 0) {
// Infinite wait.
while (!worker->HasMessageLocked()) {
int rc = pthread_cond_wait(&worker->cond_, &worker->mutex_);
if (rc != 0) break;
}
if (worker->HasMessageLocked()) {
result = worker->GetMessageNotEmptyLocked(message);
}
} else if (timeout_ms > 0) {
// Timed wait.
struct timespec ts;
struct timeval tp;
gettimeofday(&tp, NULL);
ts.tv_sec = tp.tv_sec;
ts.tv_nsec = tp.tv_usec * 1000 + timeout_ms * 1000000;
ts.tv_sec += ts.tv_nsec / 1000000000;
ts.tv_nsec %= 1000000000;
while (!worker->HasMessageLocked()) {
int rc = pthread_cond_timedwait(&worker->cond_, &worker->mutex_, &ts);
if (rc != 0) break;
}
if (worker->HasMessageLocked()) {
result = worker->GetMessageNotEmptyLocked(message);
}
}
worker->blocked_on_message_ = nullptr;
worker->blocked_message_used_ = false;
}
pthread_mutex_unlock(&worker->mutex_);
return result;
}
#ifdef __cplusplus
} // extern "C"
#endif
samples/concurrent/src/main/cpp/MessageChannel.h
-63
View File
@@ -1,63 +0,0 @@
/*
* 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.
*/
#ifndef MESSAGE_CHANNEL_H
#define MESSAGE_CHANNEL_H
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
typedef int32_t worker_id_t;
typedef int32_t message_kind_t;
typedef struct {
worker_id_t source_;
worker_id_t destination_;
message_kind_t kind_;
void* data_;
int64_t data_size_;
int64_t data_capacity_;
} Message;
#define INVALID_WORKER -1
// Creates new worker, return its id, or INVALID_WORKER if can not.
worker_id_t CreateWorker(const char* where, int argc, const char** argv);
// Gets id of the current worker.
worker_id_t CurrentWorker();
// Create message.
Message* CreateMessage(int64_t data_size);
void ReleaseMessage(Message* message);
// Returns 0 is message was delivered to 'destination' event queue,
// and -1 if destination is invalid or cannot accept more messages.
int SendMessage(worker_id_t destination, const Message* message);
// Gets next message, returns -1 if no message arrived until timeout
// expired. 'timeout_ms' can have following values:
// * > 0 - number of milliseconds to wait
// * == 0 - return immediately if there's a message already
// * < 0 - wait forever, until event arrives or program terminates
int GetMessage(Message* message, int timeout_ms);
#ifdef __cplusplus
} // extern "C"
#endif
#endif // MESSAGE_CHANNEL_H
samples/concurrent/src/main/kotlin/Concurrent.kt
-108
View File
@@ -1,108 +0,0 @@
/*
* 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.
*/
import kotlinx.cinterop.*
import MessageChannel.*
val nameToWorker = mapOf("worker1" to ::Worker1, "worker2" to ::Worker2)
fun StartWorker(target: String, name: String, vararg args: String): worker_id_t {
return memScoped {
val workerArgs = arrayOf(name, *args)
CreateWorker(target, workerArgs.size, workerArgs.map { it.cstr.getPointer(memScope) }.toCValues())
}
}
fun CheckWorker(args: Array<String>) : Boolean {
if (args.size == 0) return false
val handler = nameToWorker[args[0]]
if (handler == null) return false
handler(args)
return true
}
fun Worker1(args: Array<String>) {
println("I am Worker1 passed ${args[1]}")
memScoped {
val message = CreateMessage(10)!!
var main_thread : Int
while (true) {
val result = GetMessage(message, -1)
if (result == 0) {
main_thread = message.pointed.source_
println("${args[1]} got message ${message.pointed.kind_} ${message.pointed.source_}->${message.pointed.destination_}")
if (message.pointed.kind_ == 42) break
}
}
message.pointed.kind_ = 1
SendMessage(main_thread, message)
}
}
fun Worker2(args: Array<String>) {
println("I am Worker2 passed ${args[1]}")
memScoped {
val message = CreateMessage(10)!!
var main_thread : Int
while (true) {
val result = GetMessage(message, -1)
if (result == 0) {
main_thread = message.pointed.source_
println("${args[1]} got message ${message.pointed.kind_} ${message.pointed.source_}->${message.pointed.destination_}")
if (message.pointed.kind_ == 42) break
}
}
message.pointed.kind_ = 2
SendMessage(main_thread, message)
}
}
fun main(args: Array<String>) {
val id = if (args.size > 0) args[0] else "main"
println("Hi from $id")
if (CheckWorker(args)) return
memScoped {
val worker1 = StartWorker("thread", "worker1", "Foo")
val worker2 = StartWorker("thread", "worker2", "Bar")
val message1 = CreateMessage(20)!!
val message2 = CreateMessage(20)!!
message1.pointed.kind_ = 1239
message2.pointed.kind_ = 1566
for (i in 1 .. 200) {
SendMessage(worker1, message1)
message1.pointed.kind_++
SendMessage(worker2, message2)
message2.pointed.kind_--
}
// Send termination message.
message1.pointed.kind_ = 42
SendMessage(worker1, message1)
message2.pointed.kind_ = 42
SendMessage(worker2, message2)
// Wait for termination acknowledgment.
for (i in 1 .. 2) {
val result = GetMessage(message1, 1000)
if (result == 0) {
println("main got pingpack ${message1.pointed.kind_}")
}
}
}
}
samples/settings.gradle
-1
View File
@@ -1,4 +1,3 @@
// TODO: Uncomment
include ':csvparser'
include ':gitchurn'
include ':gtk'