text:
#include "assembly_graph_builder.h"

#include "assembly_graph_store.h"
#include "assembly_sliding_window.h"
#include "assembly_block_classifier.h"

#include "assembly_arc_kernel.h"
#include "assembly_arc_classifier.h"

#include <genomics/helpers/command.h>

#include <core/system/debugger.h>
#include <core/system/command.h>

#include <core/structures/string.h>

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include <inttypes.h>
#include <math.h>

/*
 * Constructor and destructor
 */
void biosal_assembly_graph_builder_init(struct thorium_actor *self);
void biosal_assembly_graph_builder_destroy(struct thorium_actor *self);

/*
 * The main handler
 */
void biosal_assembly_graph_builder_receive(struct thorium_actor *self, struct thorium_message *message);

/*
 * Other handlers
 */
void biosal_assembly_graph_builder_ask_to_stop(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_start(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_set_producers(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_spawn_reply(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_spawn_reply_graph_store_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_spawn_reply_window_manager(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_set_script_reply_store_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_start_reply_store_manager(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_set_script_reply_window_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_start_reply_window_manager(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_set_script_reply_classifier_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_start_reply_classifier_manager(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_configure(struct thorium_actor *self);
void biosal_assembly_graph_builder_set_kmer_reply(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_connect_actors(struct thorium_actor *self);

void biosal_assembly_graph_builder_set_consumers_reply(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_verify(struct thorium_actor *self);
void biosal_assembly_graph_builder_set_consumer_reply(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_tell_source(struct thorium_actor *self);
void biosal_assembly_graph_builder_set_producer_reply(struct thorium_actor *self, struct thorium_message *message);
int biosal_assembly_graph_builder_get_kmer_length(struct thorium_actor *self);

void biosal_assembly_graph_builder_notify_reply(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_control_complexity(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_get_entry_count_reply(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_set_consumer_reply_graph_stores(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_set_consumer_reply_windows(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_set_expected_message_count_reply(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_notify_from_distribution(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_spawn_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_start_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_set_script_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message);

/*
 * For the manager of arc classifiers.
 */
void biosal_assembly_graph_builder_start_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_set_script_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_spawn_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message);

/*
 * Configure kmer length for arcs.
 */

void biosal_assembly_graph_builder_set_kmer_reply_arcs(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_configure_arc_actors(struct thorium_actor *self, struct thorium_message *message);

/*
 * Functions to verify arcs for high quality.
 */
void biosal_assembly_graph_builder_verify_arcs(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_verify_arc_kernels(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_notify_reply_arc_kernels(struct thorium_actor *self, struct thorium_message *message);

void biosal_assembly_graph_builder_get_summary_reply(struct thorium_actor *self, struct thorium_message *message);
void biosal_assembly_graph_builder_get_producers_for_work_stealing(struct thorium_actor *self, struct core_vector *producers_for_work_stealing,
                int current_index);
void biosal_assembly_graph_builder_set_actors_reply_store_manager(struct thorium_actor *self, struct thorium_message *message);

struct thorium_script biosal_assembly_graph_builder_script = {
    .identifier = SCRIPT_ASSEMBLY_GRAPH_BUILDER,
    .name = "biosal_assembly_graph_builder",
    .description = "This builder implements a distributed actor algorithm for building an assembly graph",
    .version = "Cool version",
    .author = "Sebastien Boisvert, Argonne National Laboratory",
    .init = biosal_assembly_graph_builder_init,
    .destroy = biosal_assembly_graph_builder_destroy,
    .receive = biosal_assembly_graph_builder_receive,
    .size = sizeof(struct biosal_assembly_graph_builder)
};

void biosal_assembly_graph_builder_init(struct thorium_actor *self)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    core_timer_init(&concrete_self->timer);
    core_timer_init(&concrete_self->vertex_timer);
    core_timer_init(&concrete_self->arc_timer);

    concrete_self->completed_sliding_windows = 0;
    concrete_self->doing_arcs = 0;

    core_vector_init(&concrete_self->spawners, sizeof(int));
    core_vector_init(&concrete_self->sequence_stores, sizeof(int));
    core_vector_init(&concrete_self->block_classifiers, sizeof(int));

    thorium_actor_add_action(self, ACTION_START, biosal_assembly_graph_builder_start);
    thorium_actor_add_action(self, ACTION_SET_PRODUCERS, biosal_assembly_graph_builder_set_producers);
    thorium_actor_add_action(self, ACTION_SET_PRODUCER_REPLY, biosal_assembly_graph_builder_set_producer_reply);
    thorium_actor_add_action(self, ACTION_ASK_TO_STOP, biosal_assembly_graph_builder_ask_to_stop);
    thorium_actor_add_action(self, ACTION_SPAWN_REPLY, biosal_assembly_graph_builder_spawn_reply);
    thorium_actor_add_action(self, ACTION_SET_KMER_LENGTH_REPLY, biosal_assembly_graph_builder_set_kmer_reply);
    thorium_actor_add_action(self, ACTION_NOTIFY_REPLY, biosal_assembly_graph_builder_notify_reply);
    thorium_actor_add_action(self, ACTION_ASSEMBLY_GRAPH_BUILDER_CONTROL_COMPLEXITY,
                    biosal_assembly_graph_builder_control_complexity);

    thorium_actor_add_action(self, ACTION_SET_CONSUMER_REPLY, biosal_assembly_graph_builder_set_consumer_reply);
    thorium_actor_add_action(self, ACTION_SET_CONSUMERS_REPLY, biosal_assembly_graph_builder_set_consumers_reply);
    thorium_actor_add_action(self, ACTION_STORE_GET_ENTRY_COUNT_REPLY,
                    biosal_assembly_graph_builder_get_entry_count_reply);

    concrete_self->manager_for_graph_stores = THORIUM_ACTOR_NOBODY;
    core_vector_init(&concrete_self->graph_stores, sizeof(int));

    concrete_self->manager_for_classifiers = THORIUM_ACTOR_NOBODY;

    concrete_self->manager_for_windows = THORIUM_ACTOR_NOBODY;
    core_vector_init(&concrete_self->sliding_windows, sizeof(int));

    concrete_self->coverage_distribution = THORIUM_ACTOR_NOBODY;

    concrete_self->manager_for_arc_kernels = THORIUM_ACTOR_NOBODY;
    core_vector_init(&concrete_self->arc_kernels, sizeof(int));

    concrete_self->manager_for_arc_classifiers = THORIUM_ACTOR_NOBODY;
    core_vector_init(&concrete_self->arc_classifiers, sizeof(int));

    concrete_self->configured_graph_stores = 0;
    concrete_self->configured_sliding_windows = 0;
    concrete_self->configured_block_classifiers = 0;

    concrete_self->actors_with_kmer_length = 0;

    concrete_self->total_kmer_count = 0;
    concrete_self->notified_windows = 0;

    concrete_self->synchronized_graph_stores = 0;
    concrete_self->actual_kmer_count = 0;

    concrete_self->expected_arc_count = 0;

    biosal_assembly_graph_summary_init(&concrete_self->graph_summary);
}

void biosal_assembly_graph_builder_destroy(struct thorium_actor *self)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    core_vector_destroy(&concrete_self->spawners);
    core_vector_destroy(&concrete_self->sequence_stores);

    concrete_self->manager_for_graph_stores = THORIUM_ACTOR_NOBODY;
    core_vector_destroy(&concrete_self->graph_stores);

    concrete_self->manager_for_classifiers = THORIUM_ACTOR_NOBODY;
    core_vector_destroy(&concrete_self->block_classifiers);

    concrete_self->manager_for_windows= THORIUM_ACTOR_NOBODY;
    core_vector_destroy(&concrete_self->sliding_windows);

    concrete_self->manager_for_arc_kernels = THORIUM_ACTOR_NOBODY;
    core_vector_destroy(&concrete_self->arc_kernels);

    concrete_self->manager_for_arc_classifiers = THORIUM_ACTOR_NOBODY;
    core_vector_destroy(&concrete_self->arc_classifiers);

    core_timer_destroy(&concrete_self->timer);
    core_timer_destroy(&concrete_self->vertex_timer);
    core_timer_destroy(&concrete_self->arc_timer);

    biosal_assembly_graph_summary_destroy(&concrete_self->graph_summary);
}

void biosal_assembly_graph_builder_receive(struct thorium_actor *self, struct thorium_message *message)
{
    int action = thorium_message_action(message);
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    thorium_actor_take_action(self, message);

    LOG("receive() %d:%d", thorium_message_source(message),
                    thorium_message_get_identifier(message));

    if (action == ACTION_SET_PRODUCER) {

        /*
         * This chunk is just a test for the _send_then() code.
         */
        thorium_actor_log(self,
                        "XXX graph builder received ACTION_SET_PRODUCER, responding with ACTION_SET_PRODUCER_REPLY.\n");

        thorium_actor_send_reply_empty(self, ACTION_SET_PRODUCER_REPLY);
    } else if (action == ACTION_GET_MANAGER) {

        REPLY(1, ACTION_GET_MANAGER, TYPE_INT, concrete_self->manager_for_graph_stores);
    }
}

void biosal_assembly_graph_builder_ask_to_stop(struct thorium_actor *self, struct thorium_message *message)
{
    int name;
    struct biosal_assembly_graph_builder *concrete_self;

    name = thorium_actor_name(self);
    concrete_self = thorium_actor_concrete_actor(self);

    thorium_actor_log(self, "builder/%d dies\n", name);

    thorium_actor_ask_to_stop(self, message);

    /*
     * Stop graph stores
     */

    thorium_actor_send_empty(self, concrete_self->manager_for_graph_stores,
                    ACTION_ASK_TO_STOP);
}

void biosal_assembly_graph_builder_start(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    struct biosal_assembly_graph_builder *concrete_self;
    int source;
    int spawner;

    source = thorium_message_source(message);
    concrete_self = thorium_actor_concrete_actor(self);

    if (source != concrete_self->source) {
        return;
    }

    core_timer_start(&concrete_self->timer);
    core_timer_start(&concrete_self->vertex_timer);

    buffer = thorium_message_buffer(message);

    core_vector_unpack(&concrete_self->spawners, buffer);

    thorium_actor_log(self, "%s/%d received a urgent request from source %d to build an assembly graph using %d sequence stores (producers) and %d spawners\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    source,
                    (int)core_vector_size(&concrete_self->sequence_stores),
                    (int)core_vector_size(&concrete_self->spawners));

    spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

    /*
     * Spawn the manager for graph stores
     */
    concrete_self->manager_for_graph_stores = THORIUM_ACTOR_SPAWNING_IN_PROGRESS;

    LOG("XXX debug %d send to spawner %d\n", NAME(), spawner);

    thorium_actor_send_then_int(self, spawner, ACTION_SPAWN, SCRIPT_MANAGER,
                    biosal_assembly_graph_builder_spawn_reply_graph_store_manager);
}

void biosal_assembly_graph_builder_spawn_reply_graph_store_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    LOG("XXX callback, src %d", SRC(message));

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Configure the manager for graph stores
     * Make sure that this callback is only used once.
     */

    CORE_DEBUGGER_ASSERT(concrete_self->manager_for_graph_stores == THORIUM_ACTOR_SPAWNING_IN_PROGRESS);

    thorium_message_unpack_int(message, 0, &concrete_self->manager_for_graph_stores);

    thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_SCRIPT_REPLY,
                    biosal_assembly_graph_builder_set_script_reply_store_manager,
                    concrete_self->manager_for_graph_stores);

    thorium_actor_add_action_with_source(self, ACTION_START_REPLY,
                    biosal_assembly_graph_builder_start_reply_store_manager,
                    concrete_self->manager_for_graph_stores);

    thorium_actor_send_int(self, concrete_self->manager_for_graph_stores, ACTION_MANAGER_SET_SCRIPT,
                        SCRIPT_ASSEMBLY_GRAPH_STORE);
}

/*
 * find some sort of way to reduce the number of lines here
 * using a new Thorium API call.
 *
 * Solutions:
 *
 * thorium_actor_add_action_with_condition
 * thorium_actor_add_action_with_source_and_condition
 */
void biosal_assembly_graph_builder_spawn_reply(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    thorium_actor_log(self, "DEBUG biosal_assembly_graph_builder_spawn_reply\n");

    /*
     * Configure the manager for block classifiers
     */
    if (concrete_self->manager_for_classifiers == THORIUM_ACTOR_NOBODY) {

        thorium_message_unpack_int(message, 0, &concrete_self->manager_for_classifiers);

        thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_SCRIPT_REPLY,
                        biosal_assembly_graph_builder_set_script_reply_classifier_manager,
                        concrete_self->manager_for_classifiers);

        thorium_actor_add_action_with_source(self, ACTION_START_REPLY,
                        biosal_assembly_graph_builder_start_reply_classifier_manager,
                        concrete_self->manager_for_classifiers);

        thorium_actor_send_int(self, concrete_self->manager_for_classifiers, ACTION_MANAGER_SET_SCRIPT,
                        SCRIPT_ASSEMBLY_BLOCK_CLASSIFIER);

    } else if (concrete_self->coverage_distribution == THORIUM_ACTOR_NOBODY) {

        thorium_message_unpack_int(message, 0, &concrete_self->coverage_distribution);

        thorium_actor_add_action_with_source(self, ACTION_SET_EXPECTED_MESSAGE_COUNT_REPLY,
                            biosal_assembly_graph_builder_set_expected_message_count_reply,
                        concrete_self->coverage_distribution);

        thorium_actor_add_action_with_source(self, ACTION_NOTIFY,
                        biosal_assembly_graph_builder_notify_from_distribution,
                        concrete_self->coverage_distribution);

        thorium_actor_send_int(self, concrete_self->coverage_distribution,
                        ACTION_SET_EXPECTED_MESSAGE_COUNT, (int)core_vector_size(&concrete_self->graph_stores));

    } else {

        thorium_actor_log(self, "Warning: unknown state when receiveing ACTION_SPAWN_REPLY\n");
    }
}

void biosal_assembly_graph_builder_set_expected_message_count_reply(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_actor_add_action_with_sources(self, ACTION_SET_CONSUMER_REPLY,
                    biosal_assembly_graph_builder_set_consumer_reply_graph_stores,
                    &concrete_self->graph_stores);

    /*
     * Link the graph stores to the coverage distribution
     */
    thorium_actor_send_range_int(self, &concrete_self->graph_stores,
                    ACTION_SET_CONSUMER,
                    concrete_self->coverage_distribution);
}

void biosal_assembly_graph_builder_set_script_reply_store_manager(struct thorium_actor *self, struct thorium_message *message)
{
        /*
    struct biosal_assembly_graph_builder *concrete_self;
    */

#if 0
    int actor_count;
#endif
    int actors_per_spawner;
    int source;

    /*
     * Verify is the number of actors is too high.
     */
    actors_per_spawner = biosal_assembly_graph_store_get_store_count_per_node(self);
    /*
    actor_count = node_count * actors_per_spawner;
    */
    source = thorium_message_source(message);

    /*
    concrete_self = thorium_actor_concrete_actor(self);
    */

    thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_ACTORS_PER_SPAWNER_REPLY,
                    biosal_assembly_graph_builder_set_actors_reply_store_manager,
                    source);

    thorium_actor_send_reply_int(self, ACTION_MANAGER_SET_ACTORS_PER_SPAWNER,
                    actors_per_spawner);
}

void biosal_assembly_graph_builder_set_actors_reply_store_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    thorium_actor_send_reply_vector(self, ACTION_START, &concrete_self->spawners);
}

void biosal_assembly_graph_builder_start_reply_store_manager(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    int spawner;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    buffer = thorium_message_buffer(message);

    core_vector_unpack(&concrete_self->graph_stores, buffer);

    thorium_actor_log(self, "%s/%d has %d graph stores for assembly\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    (int)core_vector_size(&concrete_self->graph_stores));

    /* Spawn the manager for the sliding windows manager
     */

    spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

    /*
     * Register event to happen before
     * sending message (which generates an event)
     */
    concrete_self->manager_for_windows = THORIUM_ACTOR_SPAWNING_IN_PROGRESS;

    thorium_actor_add_action_with_condition(self,
                    ACTION_SPAWN_REPLY,
                    biosal_assembly_graph_builder_spawn_reply_window_manager,
                    &concrete_self->manager_for_windows,
                    THORIUM_ACTOR_SPAWNING_IN_PROGRESS);

    thorium_actor_send_int(self, spawner, ACTION_SPAWN,
                    SCRIPT_MANAGER);
}

void biosal_assembly_graph_builder_spawn_reply_window_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Configure the manager
     * for sliding windows
     */

    thorium_message_unpack_int(message, 0, &concrete_self->manager_for_windows);

    thorium_actor_log(self, "%s/%d says that the manager for windows is %d\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    concrete_self->manager_for_windows);

    thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_SCRIPT_REPLY,
                    biosal_assembly_graph_builder_set_script_reply_window_manager,
                    concrete_self->manager_for_windows);

    thorium_actor_add_action_with_source(self, ACTION_START_REPLY,
                    biosal_assembly_graph_builder_start_reply_window_manager,
                    concrete_self->manager_for_windows);

    thorium_actor_send_int(self, concrete_self->manager_for_windows, ACTION_MANAGER_SET_SCRIPT,
                    SCRIPT_ASSEMBLY_SLIDING_WINDOW);
}

void biosal_assembly_graph_builder_set_producers(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    struct biosal_assembly_graph_builder *concrete_self;
    int source;

    source = thorium_message_source(message);

    concrete_self = thorium_actor_concrete_actor(self);
    concrete_self->source = source;

    buffer = thorium_message_buffer(message);

    thorium_actor_log(self, "%s/%d received a %d sequence stores from source %d\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    (int)core_vector_size(&concrete_self->sequence_stores),
                    concrete_self->source);

    /*
     * Unpack sequence stores in the concrete actor.
     */
    core_vector_unpack(&concrete_self->sequence_stores, buffer);

    thorium_actor_send_reply_empty(self, ACTION_SET_PRODUCERS_REPLY);
}

void biosal_assembly_graph_builder_set_script_reply_window_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    thorium_actor_send_reply_vector(self, ACTION_START, &concrete_self->spawners);
}

void biosal_assembly_graph_builder_start_reply_window_manager(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    struct biosal_assembly_graph_builder *concrete_self;
    int spawner;

    concrete_self = thorium_actor_concrete_actor(self);

    buffer = thorium_message_buffer(message);

    /*
     * Unpack the actor names for the sliding
     * windows
     */
    core_vector_unpack(&concrete_self->sliding_windows, buffer);

    thorium_actor_add_action_with_sources(self, ACTION_SET_CONSUMER_REPLY,
                    biosal_assembly_graph_builder_set_consumer_reply_windows,
                    &concrete_self->sliding_windows);

    thorium_actor_log(self, "%s/%d has %d sliding windows for assembly\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    (int)core_vector_size(&concrete_self->sliding_windows));

    spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

    thorium_actor_send_int(self, spawner, ACTION_SPAWN,
                    SCRIPT_MANAGER);
}

void biosal_assembly_graph_builder_set_script_reply_classifier_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    thorium_actor_send_reply_vector(self, ACTION_START, &concrete_self->spawners);
}

void biosal_assembly_graph_builder_start_reply_classifier_manager(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    buffer = thorium_message_buffer(message);
    core_vector_unpack(&concrete_self->block_classifiers, buffer);

    thorium_actor_log(self, "%s/%d has %d block classifiers for assembly\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    (int)core_vector_size(&concrete_self->block_classifiers));

    /*
     * Configure the graph builder now.
     */
    biosal_assembly_graph_builder_configure(self);
}

void biosal_assembly_graph_builder_configure(struct thorium_actor *self)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Set the default kmer length
     */
    concrete_self->kmer_length = biosal_assembly_graph_builder_get_kmer_length(self);

#if 0
    thorium_actor_log(self, "EXAMINE: before configuring kmer length\n");
    thorium_actor_print(self);
#endif

    thorium_actor_log(self, "%s/%d configures the kmer length (%d) for the actor computation\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self),
                    concrete_self->kmer_length);
    /*
     * set the kmer length for graph stores, sliding windows, and block classifiers
     */

    thorium_actor_send_range_int(self, &concrete_self->graph_stores, ACTION_SET_KMER_LENGTH,
                        concrete_self->kmer_length);

    thorium_actor_send_range_int(self, &concrete_self->sliding_windows, ACTION_SET_KMER_LENGTH,
                        concrete_self->kmer_length);

    thorium_actor_send_range_int(self, &concrete_self->block_classifiers, ACTION_SET_KMER_LENGTH,
                        concrete_self->kmer_length);

#if 0
    /*
     * There will be a response for this.
     */
    thorium_actor_log(self, "EXAMINE: after configuring kmer length\n");
    thorium_actor_print(self);
#endif
}

void biosal_assembly_graph_builder_set_kmer_reply(struct thorium_actor *self, struct thorium_message *message)
{
    int expected;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->actors_with_kmer_length;

    expected = 0;
    expected += core_vector_size(&concrete_self->graph_stores);
    expected += core_vector_size(&concrete_self->sliding_windows);
    expected += core_vector_size(&concrete_self->block_classifiers);

    if (concrete_self->actors_with_kmer_length % 100 == 0) {

#ifdef EXAMINE_GRAPH_BUILDER
        thorium_actor_log(self, "EXAMINE: progress SET_KMER_LENGTH %d/%d\n",
                        concrete_self->actors_with_kmer_length,
                        expected);
        thorium_actor_print(self);
#endif
    }

    /*
     * Verify if all actors are properly configured
     */
    if (concrete_self->actors_with_kmer_length == expected) {

        thorium_actor_log(self, "EXAMINE: configured kmer length\n");

#if 0
        thorium_actor_print(self);
#endif

        thorium_actor_log(self, "%s/%d configured (%d actors) the kmer length value for sliding windows, block classifiers and graph stores\n",
                        thorium_actor_script_name(self),
                        thorium_actor_name(self),
                        expected);

        biosal_assembly_graph_builder_connect_actors(self);
    }
}

void biosal_assembly_graph_builder_connect_actors(struct thorium_actor *self)
{
    int i;
    int producer;
    int consumer;
    struct biosal_assembly_graph_builder *concrete_self;
    struct core_vector producers_for_work_stealing;
    struct core_memory_pool *ephemeral_memory;

    concrete_self = thorium_actor_concrete_actor(self);
    ephemeral_memory = thorium_actor_get_ephemeral_memory(self);

    core_vector_init(&producers_for_work_stealing, sizeof(int));
    core_vector_set_memory_pool(&producers_for_work_stealing, ephemeral_memory);

    /*
     * sequence store ===> sliding window ===> block classifier ===> graph store
     */

    thorium_actor_log(self, "%s/%d connects actors\n",
            thorium_actor_script_name(self),
            thorium_actor_name(self));

#ifdef EXAMINE_GRAPH_BUILDER
    thorium_actor_log(self, "EXAMINE: connecting actors\n");
    thorium_actor_print(self);
#endif

    thorium_actor_add_action_with_sources(self, ACTION_SET_PRODUCERS_FOR_WORK_STEALING_REPLY,
                    biosal_assembly_graph_builder_set_consumer_reply_windows,
                    &concrete_self->sliding_windows);

    for (i = 0; i < core_vector_size(&concrete_self->sliding_windows); ++i) {

        biosal_assembly_graph_builder_get_producers_for_work_stealing(self, &producers_for_work_stealing, i);

        producer = core_vector_at_as_int(&concrete_self->sliding_windows, i);
        thorium_actor_send_vector(self, producer, ACTION_SET_PRODUCERS_FOR_WORK_STEALING,
                        &producers_for_work_stealing);
    }

#ifdef EXAMINE_GRAPH_BUILDER
    thorium_actor_log(self, "EXAMINE: after sending producers for work stealing\n");
    thorium_actor_print(self);
#endif

    for (i = 0; i < core_vector_size(&concrete_self->sliding_windows); i++) {

        producer = core_vector_at_as_int(&concrete_self->sliding_windows, i);
        consumer = core_vector_at_as_int(&concrete_self->block_classifiers, i);

        /*
         * Also configure a bunch of alternate producer for the
         * consumer.
         */

        /* set the consumer for sliding window
         */
        thorium_actor_send_int(self, producer, ACTION_SET_CONSUMER,
                        consumer);

        thorium_actor_log(self, "DEBUG neural LINK %d -> %d\n",
                        producer, consumer);

        /*
         * set the consumers for each classifier
         */

        thorium_actor_log(self, "DEBUG sending %d store names to classifier %d\n",
                        (int)core_vector_size(&concrete_self->graph_stores),
                        consumer);
    }

#ifdef EXAMINE_GRAPH_BUILDER
    thorium_actor_log(self, "EXAMINE: before SET_CONSUMERS\n");
    thorium_actor_print(self);
#endif

    thorium_actor_send_range_vector(self, &concrete_self->block_classifiers,
                    ACTION_SET_CONSUMERS,
                        &concrete_self->graph_stores);

    core_vector_destroy(&producers_for_work_stealing);

#ifdef EXAMINE_GRAPH_BUILDER
    thorium_actor_log(self, "EXAMINE: after connecting actors\n");
    thorium_actor_print(self);
#endif
}

void biosal_assembly_graph_builder_set_consumer_reply(struct thorium_actor *self, struct thorium_message *message)
{
}

void biosal_assembly_graph_builder_set_consumer_reply_graph_stores(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->configured_graph_stores;

    if (concrete_self->configured_graph_stores == core_vector_size(&concrete_self->graph_stores)) {

        /*
         * ABCD
         */

        thorium_actor_send_range_empty(self, &concrete_self->graph_stores,
                        ACTION_PUSH_DATA);
    }
}

void biosal_assembly_graph_builder_set_consumer_reply_windows(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->configured_sliding_windows;

#ifdef BIOSAL_ASSEMBLY_GRAPH_BUILDER_DEBUG
    thorium_actor_log(self, "DEBUG windows UP %d\n",
        concrete_self->configured_sliding_windows);
#endif

    biosal_assembly_graph_builder_verify(self);
}

void biosal_assembly_graph_builder_verify(struct thorium_actor *self)
{
    struct biosal_assembly_graph_builder *concrete_self;
    int i;
    int producer;
    int consumer;

    concrete_self = thorium_actor_concrete_actor(self);

    /* Verify if the system is ready to roll
     */
    if (concrete_self->configured_sliding_windows < 2 * core_vector_size(&concrete_self->sliding_windows)) {
        return;
    }

    if (concrete_self->configured_block_classifiers < core_vector_size(&concrete_self->block_classifiers)) {
        return;
    }

    thorium_actor_log(self, "%s/%d is ready to build the graph\n",
            thorium_actor_script_name(self),
            thorium_actor_name(self));

#ifdef EXAMINE_GRAPH_BUILDER
    thorium_actor_log(self, "EXAMINE: ready to build\n");
    thorium_actor_print(self);
#endif

    /* Set the producer for every sliding window.
     */

    for (i = 0; i < core_vector_size(&concrete_self->sliding_windows); i++) {

        producer = core_vector_at_as_int(&concrete_self->sequence_stores, i);
        consumer = core_vector_at_as_int(&concrete_self->sliding_windows, i);

        thorium_actor_log(self, "CONFIGURE neural LINK store %d -> window %d\n",
                        producer, consumer);

        thorium_actor_send_int(self, consumer, ACTION_SET_PRODUCER,
                        producer);
    }
}

void biosal_assembly_graph_builder_set_consumers_reply(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->configured_block_classifiers;

#ifdef BIOSAL_ASSEMBLY_GRAPH_BUILDER_DEBUG
    thorium_actor_log(self, "DEBUG classifiers UP %d\n", concrete_self->configured_block_classifiers);
#endif

    biosal_assembly_graph_builder_verify(self);
}

void biosal_assembly_graph_builder_set_producer_reply(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->completed_sliding_windows;

    /*
     * Tell the source that the graph is ready now.
     */
    if (concrete_self->completed_sliding_windows == core_vector_size(&concrete_self->sliding_windows)) {

        thorium_actor_send_range_empty(self, &concrete_self->sliding_windows, ACTION_NOTIFY);
    }
}

void biosal_assembly_graph_builder_tell_source(struct thorium_actor *self)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Stop actors
     */
    /*
    thorium_actor_send_range_empty(self, &concrete_self->arc_kernels, ACTION_ASK_TO_STOP);
    */
    /*
     * We need to stop the arc supervisor to stop
     * arc kernels.
     * The current actor can not stop arc kernels directly.
     */

    thorium_actor_send_empty(self, concrete_self->manager_for_arc_kernels, ACTION_ASK_TO_STOP);
    thorium_actor_send_empty(self, concrete_self->manager_for_arc_classifiers, ACTION_ASK_TO_STOP);

    /*
     * Show timer
     */
    core_timer_stop(&concrete_self->arc_timer);
    core_timer_stop(&concrete_self->timer);
    core_timer_print_with_description(&concrete_self->arc_timer,
                    "Build assembly graph / Distribute arcs");
    core_timer_print_with_description(&concrete_self->timer,
                    "Build assembly graph");

    /*
     * Tell somebody about it
     */
    thorium_actor_send_vector(self, concrete_self->source, ACTION_START_REPLY,
                    &concrete_self->graph_stores);
}

int biosal_assembly_graph_builder_get_kmer_length(struct thorium_actor *self)
{
    int kmer_length;
    int argc;
    char **argv;

    argc = thorium_actor_argc(self);
    argv = thorium_actor_argv(self);

    kmer_length = biosal_command_get_kmer_length(argc, argv);

    return kmer_length;
}

void biosal_assembly_graph_builder_notify_reply(struct thorium_actor *self, struct thorium_message *message)
{
    uint64_t produced_kmer_count;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_message_unpack_uint64_t(message, 0, &produced_kmer_count);

    concrete_self->total_kmer_count += produced_kmer_count;
    ++concrete_self->notified_windows;

    /*
     * All sliding window terminated their job of extracting kmers.
     */
    if (concrete_self->notified_windows == core_vector_size(&concrete_self->sliding_windows)) {

        thorium_actor_log(self, "%s/%d : %" PRIu64 " kmers were generated during the actor computation.\n",
            thorium_actor_script_name(self),
            thorium_actor_name(self),
            concrete_self->total_kmer_count);

        /*
         * Tell all classifiers to flush now.
         *
         * It may also be required to do this later on too.
         */
        thorium_actor_send_range_empty(self, &concrete_self->block_classifiers,
                        ACTION_AGGREGATOR_FLUSH);

        /*
         * Use message passing for control.
         *
         * \see http://dspace.mit.edu/handle/1721.1/6272
         */
        thorium_actor_send_to_self_empty(self, ACTION_ASSEMBLY_GRAPH_BUILDER_CONTROL_COMPLEXITY);
    }
}

void biosal_assembly_graph_builder_control_complexity(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Tell all classifiers to flush now.
     *
     * This second call is required because if the messages are passed too slowly (or too fast),
     * the FLUSH signal can arrive before the bit messages actually get there.
     */
    thorium_actor_send_range_empty(self, &concrete_self->block_classifiers,
                    ACTION_AGGREGATOR_FLUSH);

    concrete_self->actual_kmer_count = 0;
    concrete_self->synchronized_graph_stores = 0;

    thorium_actor_send_range_empty(self, &concrete_self->graph_stores,
                    ACTION_STORE_GET_ENTRY_COUNT);

    thorium_actor_log(self, "debug biosal_assembly_graph_builder_control_complexity %d graph stores\n",
                    (int)core_vector_size(&concrete_self->graph_stores));
}

void biosal_assembly_graph_builder_get_entry_count_reply(struct thorium_actor *self, struct thorium_message *message)
{
    uint64_t kmer_count;
    struct biosal_assembly_graph_builder *concrete_self;
    int spawner;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_message_unpack_uint64_t(message, 0, &kmer_count);

    concrete_self->actual_kmer_count += kmer_count;

    ++concrete_self->synchronized_graph_stores;

    if (concrete_self->synchronized_graph_stores % 100 == 0
                    || concrete_self->synchronized_graph_stores ==
                    (int)core_vector_size(&concrete_self->graph_stores)) {

        thorium_actor_log(self, "synchronized_graph_stores %d/%d %" PRIu64 "/%" PRIu64 "\n",
                    concrete_self->synchronized_graph_stores,
                    (int)core_vector_size(&concrete_self->graph_stores),
                    concrete_self->actual_kmer_count,
                    concrete_self->total_kmer_count);
    }

    if (concrete_self->synchronized_graph_stores == core_vector_size(&concrete_self->graph_stores)) {

        thorium_actor_log(self, "graph store synchronization: actual_kmer_count %" PRIu64 " total_kmer_count %" PRIu64 "\n",
                        concrete_self->actual_kmer_count,
                        concrete_self->total_kmer_count);

        if (concrete_self->actual_kmer_count == concrete_self->total_kmer_count) {

            thorium_actor_log(self, "%s/%d synchronized with %d graph stores\n",
                            thorium_actor_script_name(self),
                            thorium_actor_name(self),
                            (int)core_vector_size(&concrete_self->graph_stores));

            /* Generate coverage distribution
             */

            spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

            thorium_actor_send_int(self, spawner, ACTION_SPAWN,
                            SCRIPT_COVERAGE_DISTRIBUTION);
        } else {
            /*
             * Otherwise, there are still some messages in transit.
             */

            thorium_actor_log(self, "got mismatch, will try again\n");

            thorium_actor_send_to_self_empty(self, ACTION_ASSEMBLY_GRAPH_BUILDER_CONTROL_COMPLEXITY);
        }
    }
}

void biosal_assembly_graph_builder_notify_from_distribution(struct thorium_actor *self, struct thorium_message *message)
{
    int spawner;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    /*
     * Kill windows and classifiers
     */

    thorium_actor_send_empty(self, concrete_self->manager_for_windows,
                    ACTION_ASK_TO_STOP);

    thorium_actor_send_empty(self, concrete_self->manager_for_classifiers,
                    ACTION_ASK_TO_STOP);

    thorium_actor_send_empty(self, concrete_self->coverage_distribution,
                    ACTION_ASK_TO_STOP);

    core_timer_stop(&concrete_self->vertex_timer);

    core_timer_start(&concrete_self->arc_timer);
    core_timer_print_with_description(&concrete_self->vertex_timer,
                    "Build assembly graph / Distribute vertices");

    /*
     * Spawn manager for arc kernels
     */

    spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

    concrete_self->manager_for_arc_kernels = THORIUM_ACTOR_SPAWNING_IN_PROGRESS;

    thorium_actor_add_action_with_condition(self, ACTION_SPAWN_REPLY,
                    biosal_assembly_graph_builder_spawn_reply_arc_kernel_manager,
                    &concrete_self->manager_for_arc_kernels,
                    THORIUM_ACTOR_SPAWNING_IN_PROGRESS);

    thorium_actor_send_int(self, spawner, ACTION_SPAWN,
                    SCRIPT_MANAGER);
}

void biosal_assembly_graph_builder_spawn_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_message_unpack_int(message, 0, &concrete_self->manager_for_arc_kernels);

    thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_SCRIPT_REPLY,
                    biosal_assembly_graph_builder_set_script_reply_arc_kernel_manager,
                    concrete_self->manager_for_arc_kernels);

    thorium_actor_send_int(self, concrete_self->manager_for_arc_kernels, ACTION_MANAGER_SET_SCRIPT,
                    SCRIPT_ASSEMBLY_ARC_KERNEL);
}

void biosal_assembly_graph_builder_set_script_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_actor_add_action_with_source(self, ACTION_START_REPLY,
                    biosal_assembly_graph_builder_start_reply_arc_kernel_manager,
                    concrete_self->manager_for_arc_kernels);

    thorium_actor_send_reply_vector(self, ACTION_START, &concrete_self->spawners);
}

void biosal_assembly_graph_builder_start_reply_arc_kernel_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;
    void *buffer;
    int spawner;

    concrete_self = thorium_actor_concrete_actor(self);
    buffer = thorium_message_buffer(message);

    core_vector_unpack(&concrete_self->arc_kernels, buffer);

    concrete_self->manager_for_arc_classifiers = THORIUM_ACTOR_SPAWNING_IN_PROGRESS;

    thorium_actor_add_action_with_condition(self, ACTION_SPAWN_REPLY,
                    biosal_assembly_graph_builder_spawn_reply_arc_classifier_manager,
                    &concrete_self->manager_for_arc_classifiers,
                    THORIUM_ACTOR_SPAWNING_IN_PROGRESS);

    spawner = thorium_actor_get_spawner(self, &concrete_self->spawners);

    thorium_actor_send_int(self, spawner, ACTION_SPAWN, SCRIPT_MANAGER);
}

void biosal_assembly_graph_builder_spawn_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_message_unpack_int(message, 0, &concrete_self->manager_for_arc_classifiers);

    /*
     * Set the script.
     */
    thorium_actor_add_action_with_source(self, ACTION_MANAGER_SET_SCRIPT_REPLY,
                    biosal_assembly_graph_builder_set_script_reply_arc_classifier_manager,
                    concrete_self->manager_for_arc_classifiers);

    thorium_actor_send_int(self, concrete_self->manager_for_arc_classifiers, ACTION_MANAGER_SET_SCRIPT,
                    SCRIPT_ASSEMBLY_ARC_CLASSIFIER);

}

void biosal_assembly_graph_builder_set_script_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_actor_add_action_with_source(self, ACTION_START_REPLY,
                    biosal_assembly_graph_builder_start_reply_arc_classifier_manager,
                    concrete_self->manager_for_arc_classifiers);

    thorium_actor_send_reply_vector(self, ACTION_START,
                    &concrete_self->spawners);
}

void biosal_assembly_graph_builder_start_reply_arc_classifier_manager(struct thorium_actor *self, struct thorium_message *message)
{
    void *buffer;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    buffer = thorium_message_buffer(message);
    core_vector_unpack(&concrete_self->arc_classifiers, buffer);

    /*
     * Configure the kmer length.
     */
    concrete_self->doing_arcs = 1;

    thorium_actor_add_action_with_condition(self, ACTION_SET_KMER_LENGTH_REPLY,
                    biosal_assembly_graph_builder_set_kmer_reply_arcs,
                    &concrete_self->doing_arcs, 1);

    concrete_self->configured_actors_for_arcs = 0;

    thorium_actor_send_range_int(self, &concrete_self->arc_kernels, ACTION_SET_KMER_LENGTH,
                    concrete_self->kmer_length);
    thorium_actor_send_range_int(self, &concrete_self->arc_classifiers, ACTION_SET_KMER_LENGTH,
                    concrete_self->kmer_length);
}

void biosal_assembly_graph_builder_set_kmer_reply_arcs(struct thorium_actor *self, struct thorium_message *message)
{
    int expected;
    struct biosal_assembly_graph_builder *concrete_self;
    int i;
    int producer;
    int consumer;
    int size;
    struct core_vector producers_for_work_stealing;
    struct core_memory_pool *ephemeral_memory;

    ephemeral_memory = thorium_actor_get_ephemeral_memory(self);
    concrete_self = thorium_actor_concrete_actor(self);

    ++concrete_self->configured_actors_for_arcs;

    expected = 0;
    expected += core_vector_size(&concrete_self->arc_kernels);
    expected += core_vector_size(&concrete_self->arc_classifiers);

    /*
     * All arc actors have a kmer length configured.
     */
    if (concrete_self->configured_actors_for_arcs < expected) {

        return;
    }

    core_vector_init(&producers_for_work_stealing, sizeof(int));
    core_vector_set_memory_pool(&producers_for_work_stealing,
                ephemeral_memory);

    thorium_actor_log(self, "%s/%d configured the kmer length for arc actors\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self));

    concrete_self->configured_actors_for_arcs = 0;

    thorium_actor_add_action_with_condition(self, ACTION_SET_CONSUMER_REPLY,
                    biosal_assembly_graph_builder_configure_arc_actors,
                    &concrete_self->doing_arcs, 1);

    size = core_vector_size(&concrete_self->arc_kernels);

    /*
     * Design:
     *
     * sequence_store => arc_kernel => arc_classifier => graph_store/0 graph_store/1 graph_store/2
     */

    /*
     * Configure action for reply to the message that sets
     * work stealing producers.
     */
    thorium_actor_add_action_with_sources(self, ACTION_SET_PRODUCERS_FOR_WORK_STEALING_REPLY,
                    biosal_assembly_graph_builder_configure_arc_actors,
                    &concrete_self->arc_kernels);

    for (i = 0; i < size; ++i) {

        biosal_assembly_graph_builder_get_producers_for_work_stealing(self, &producers_for_work_stealing, i);

        producer = core_vector_at_as_int(&concrete_self->arc_kernels, i);

        /*
         * Also set producers for work stealing too.
         */
        thorium_actor_send_vector(self, producer, ACTION_SET_PRODUCERS_FOR_WORK_STEALING,
                        &producers_for_work_stealing);
    }

    /*
     * Link kernels and classifiers
     */
    for (i = 0; i < size; i++) {

        producer = core_vector_at_as_int(&concrete_self->arc_kernels, i);
        consumer = core_vector_at_as_int(&concrete_self->arc_classifiers, i);

        thorium_actor_send_int(self, producer, ACTION_SET_CONSUMER,
                        consumer);
    }

    /*
     * Link classifiers and graph stores
     */

    thorium_actor_add_action_with_condition(self,
                    ACTION_SET_CONSUMERS_REPLY,
                    biosal_assembly_graph_builder_configure_arc_actors,
                    &concrete_self->doing_arcs, 1);

    thorium_actor_send_range_vector(self, &concrete_self->arc_classifiers,
                    ACTION_SET_CONSUMERS,
                    &concrete_self->graph_stores);

    /*
     * Reset the sequence stores now.
     */

    thorium_actor_add_action(self,
                    ACTION_RESET_REPLY,
                    biosal_assembly_graph_builder_configure_arc_actors);

    thorium_actor_send_range_empty(self, &concrete_self->sequence_stores,
                    ACTION_RESET);

    core_vector_destroy(&producers_for_work_stealing);
}

void biosal_assembly_graph_builder_configure_arc_actors(struct thorium_actor *self, struct thorium_message *message)
{
    int expected;
    int i;
    int size;
    int producer;
    int consumer;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);
    ++concrete_self->configured_actors_for_arcs;

    expected = 0;
    /* For ACTION_SET_CONSUMER_REPLY */
    expected += core_vector_size(&concrete_self->arc_kernels);
    /* For ACTION_SET_PRODUCERS_FOR_WORK_STEALING_REPLY */
    expected += core_vector_size(&concrete_self->arc_kernels);
    /* For ACTION_SET_CONSUMERS_REPLY */
    expected += core_vector_size(&concrete_self->arc_classifiers);
    /* For ACTION_RESET_REPLY */
    expected += core_vector_size(&concrete_self->sequence_stores);

    if (concrete_self->configured_actors_for_arcs % 1000 == 0
                    || concrete_self->configured_actors_for_arcs == expected) {
        thorium_actor_log(self, "PROGRESS ARC configuration %d/%d %d %d %d\n",
                    concrete_self->configured_actors_for_arcs,
                    expected,
                    (int)core_vector_size(&concrete_self->arc_kernels),
                    (int)core_vector_size(&concrete_self->arc_classifiers),
                    (int)core_vector_size(&concrete_self->sequence_stores));
    }

    if (concrete_self->configured_actors_for_arcs < expected) {
        return;
    }

    thorium_actor_log(self, "%s/%d configured the neural links for arc actors\n",
                    thorium_actor_script_name(self),
                    thorium_actor_name(self));

    concrete_self->completed_arc_kernels = 0;

    /*
     * Start arc computation
     */

    size = core_vector_size(&concrete_self->arc_kernels);

    thorium_actor_add_action_with_condition(self, ACTION_SET_PRODUCER_REPLY,
                    biosal_assembly_graph_builder_verify_arc_kernels,
                    &concrete_self->doing_arcs, 1);

    for (i = 0; i < size; i++) {
        producer = core_vector_at_as_int(&concrete_self->sequence_stores, i);
        consumer = core_vector_at_as_int(&concrete_self->arc_kernels, i);

        thorium_actor_send_int(self, consumer, ACTION_SET_PRODUCER,
                        producer);
    }
}

void biosal_assembly_graph_builder_verify_arc_kernels(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;
    int expected;

    concrete_self = thorium_actor_concrete_actor(self);

    ++concrete_self->completed_arc_kernels;

    expected = core_vector_size(&concrete_self->arc_kernels);

    if (concrete_self->completed_arc_kernels == expected) {

        concrete_self->completed_arc_kernels = 0;

        thorium_actor_add_action_with_condition(self,
                        ACTION_NOTIFY_REPLY,
                        biosal_assembly_graph_builder_notify_reply_arc_kernels,
                        &concrete_self->doing_arcs, 1);
        /*
         * Ask each kernel the amount of produced arcs
         */
        thorium_actor_send_range_empty(self, &concrete_self->arc_kernels,
                        ACTION_NOTIFY);
    }
}

void biosal_assembly_graph_builder_notify_reply_arc_kernels(struct thorium_actor *self, struct thorium_message *message)
{
    int expected;
    uint64_t arcs;
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_message_unpack_uint64_t(message, 0, &arcs);

    concrete_self->expected_arc_count += arcs;

    ++concrete_self->completed_arc_kernels;

    expected = core_vector_size(&concrete_self->arc_kernels);

    if (concrete_self->completed_arc_kernels == expected) {

        thorium_actor_log(self, "%s/%d %" PRIu64 " arcs were extracted from raw data by arc actors\n",
                        thorium_actor_script_name(self),
                        thorium_actor_name(self),
                        concrete_self->expected_arc_count);

        thorium_actor_add_action(self, ACTION_VERIFY_ARCS,
                        biosal_assembly_graph_builder_verify_arcs);

        thorium_actor_send_to_self_empty(self, ACTION_VERIFY_ARCS);
    }
}

void biosal_assembly_graph_builder_verify_arcs(struct thorium_actor *self, struct thorium_message *message)
{
    struct biosal_assembly_graph_builder *concrete_self;

    concrete_self = thorium_actor_concrete_actor(self);

    thorium_actor_add_action(self, ACTION_ASSEMBLY_GET_SUMMARY_REPLY,
                    biosal_assembly_graph_builder_get_summary_reply);

    concrete_self->ready_graph_store_count = 0;

    thorium_actor_send_range_empty(self, &concrete_self->graph_stores,
                    ACTION_ASSEMBLY_GET_SUMMARY);
}

void biosal_assembly_graph_builder_get_summary_reply(struct thorium_actor *self, struct thorium_message *message)
{
    int expected;
    struct biosal_assembly_graph_builder *concrete_self;
    void *buffer;
    struct biosal_assembly_graph_summary partial_summary;
    char *file_name;
    struct core_string file_name_string;
    char *output_directory;
    int argc;
    char **argv;

    concrete_self = thorium_actor_concrete_actor(self);
    buffer = thorium_message_buffer(message);

    biosal_assembly_graph_summary_init(&partial_summary);
    biosal_assembly_graph_summary_unpack(&partial_summary, buffer);

    biosal_assembly_graph_summary_merge(&concrete_self->graph_summary, &partial_summary);
    biosal_assembly_graph_summary_destroy(&partial_summary);

    ++concrete_self->ready_graph_store_count;

    expected = core_vector_size(&concrete_self->graph_stores);

    thorium_actor_log(self, "PROGRESS %d/%d\n", concrete_self->ready_graph_store_count,
                    expected);

    if (concrete_self->ready_graph_store_count == expected) {

            /*
        biosal_assembly_graph_summary_print(&concrete_self->graph_summary);
        */

        /*
         * Write summary file too in XML
         */

        argc = thorium_actor_argc(self);
        argv = thorium_actor_argv(self);
        output_directory = biosal_command_get_output_directory(argc, argv);
        core_string_init(&file_name_string, output_directory);
        core_string_append(&file_name_string, "/assembly_graph_summary.xml");
        file_name = core_string_get(&file_name_string);

        biosal_assembly_graph_summary_write_summary(&concrete_self->graph_summary,
                        file_name, concrete_self->kmer_length);

        core_string_destroy(&file_name_string);

        biosal_assembly_graph_builder_tell_source(self);
    }
}

void biosal_assembly_graph_builder_get_producers_for_work_stealing(struct thorium_actor *self, struct core_vector *producers_for_work_stealing,
                int current_index)
{
    struct biosal_assembly_graph_builder *concrete_self;
    int index;
    int producer;
    int store_count;
    int victim_count;
    int i;
    double probability;
    int nodes;
    int workers;
    int actors;
    double probability_of_having_an_alone_actor;
    int steps;

    /*
     * Given N actors,
     * the probability of being picked up as a victim by an actor (1 event)
     * is 1 / N.
     *
     * The probability of not being picked up is 1 - 1/N.
     *
     * If all actors picks up 1 victim, the probability of not being the victim
     * of any actor is
     *
     * (1 - 1/N) ^ ( actors * 1)
     *
     * Some examples:
     */
    /*
     * Now, given N actors, we don't want any of them to be alone.
     *
     * An actor has probability N of being selected.
     *
     *
     */
    nodes = thorium_actor_get_node_count(self);
    workers = thorium_actor_node_worker_count(self);
    actors = nodes * workers;

    /*
     * We don't want any actor alone in the process. Each actor, hopefully,
     * should be qable to get its stuff stolen by a thief.
     * But obviously that's not how life works. Everything is
     * a probability. So let's set that probability to a very small
     * value.
     *
     * 0.0001 is 0.01%.
     *
     * So the probability that at least one of the actors not being selected
     * by any other actor is very low.
     */
    probability_of_having_an_alone_actor = 0.0001;

    /* probability = 1 - pow(M_E, log(1.0 - probability_of_having_an_alone_actor) / actors);*/
    /* Otherwise, M_E can be obtained with:
     * double e = exp(1.0);
     */

    /*
     * Suppose that p is the probability that 1 actor be the victim of no one.
     * (1 - p) is then the probability of being a victim.
     *
     * (1 - p) ^ N, where N is the number of actors, is the probability that every actor
     * is a victim.
     *
     * q is the probability that at least 1 actor is not the victim of anyone.
     * To compute q, we need the probability that every actor is the victim of at
     * least an actor.
     *
     * In the relationship is:
     *
     * probability_of_having_an_alone_actor = 1.0 - pow(1.0 - probability, actors)
     *
     * "pow(1.0 - probability, actors)" represents the probability of having all
     * actors being the victim of at least 1 actor.
     *
     * The complement event is "at least one actor is not the victime of anyone.
     *
     * q = 1 - (1 - p) ^ N
     *
     * Let's isolate p.
     *
     * (1 - p) ^ N = 1 - q
     * (1 - p) = (1 - q) ^ (1 / N)
     * 1 - p = (1 - q) ^ (1 / N)
     * 1 - (1 - q) ^ (1 / N) = p
     * p = 1 - (1 - q) ^ (1 / N) ****
     */

    probability = 1 - pow(1.0 - probability_of_having_an_alone_actor, 1.0 / actors);

#ifdef DISPLAY_WORK_STEALING_STATISTICS
    thorium_actor_log(self, "PROBABILITY p of having an actor not selected by anyone %f, given %d actors\n",
                    probability_of_having_an_alone_actor, actors);

    thorium_actor_log(self, "PROBABILITY p for an actor to not being selected by anyone: %f\n",
                    probability);
#endif

    /*
     * irb(main):011:0> victims = Math.log(1.0/(5*actors))/(actors*Math.log(1 - 1.0/actors))
     */

    victim_count = log(probability) / (actors * log(1 - (1.0 / actors)));

#ifdef DISPLAY_WORK_STEALING_STATISTICS
#endif
    thorium_actor_log(self, "WORK_STEALING DEBUG victim count is %d\n", victim_count);

    /*
     * Algorithm:
     * each actor picks a constant number of peers.
     */

    concrete_self = thorium_actor_concrete_actor(self);

    store_count = core_vector_size(&concrete_self->sequence_stores);

    /*
     * Gather the list of alternate producers.
     */
    core_vector_clear(producers_for_work_stealing);

#ifdef __bDISABLEgq__
    /*
     * Disable this right now
     * because sequence stores are generating very big messages.
     *
     * There is not enough memory on Blue Gene/Q per node.
     */
    victim_count = 0;
#endif

    /*
     * There is the same number of sliding_windows and sequence_stores.
     */
    for (i = 0; i < victim_count; ++i) {
        index = current_index;

        /*
         * Avoid infinite loops if random numbers are very bad.
         */
        steps = 512;

        while (index == current_index
                        && --steps) {
            index = thorium_actor_get_random_number(self) % store_count;
        }

        producer = core_vector_at_as_int(&concrete_self->sequence_stores, index);
        core_vector_push_back(producers_for_work_stealing, &producer);
    }
}


text:
import { addStrings } from 'meteor/vulcan:core';

addStrings('en', {
  'newsletter.error_invalid_email': `Sorry, that doesn't look like a valid email.`,
  'newsletter.error_already_subscribed': `Sorry, it looks like you're already subscribed to the list.`,
  'newsletter.error_has_unsubscribed': `Sorry, it looks like you've previously unsubscribed from the list, and we're not able to re-subscribe you automatically.`,
  'newsletter.error_subscription_failed': `Sorry, your subscription failed ({message}).`,
});

redpajama_set_name: RedPajamaGithub


text:
package supervisor

import (
	"os"
)

// SignalTask holds needed parameters to signal a container
type SignalTask struct {
	baseTask
	ID     string
	PID    string
	Signal os.Signal
}

func (s *Supervisor) signal(t *SignalTask) error {
	i, ok := s.containers[t.ID]
	if !ok {
		return ErrContainerNotFound
	}
	processes, err := i.container.Processes()
	if err != nil {
		return err
	}
	for _, p := range processes {
		if p.ID() == t.PID {
			return p.Signal(t.Signal)
		}
	}
	return ErrProcessNotFound
}


text:
// Copyright (c) 2015-2020 Vladimir Schneider <vladimir.schneider@gmail.com> Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.

package com.vladsch.md.nav.psi.manipulator;

import com.intellij.openapi.util.TextRange;
import com.intellij.util.IncorrectOperationException;
import com.vladsch.md.nav.psi.element.MdInlineCode;
import org.jetbrains.annotations.NotNull;

public class MdInlineCodeManipulator implements MdElementManipulator<MdInlineCode> {
    /**
     * Changes the element's text to a new value
     *
     * @param element    element to be changed
     * @param range      range within the element
     * @param newContent new element text
     *
     * @return changed element
     *
     * @throws IncorrectOperationException if something goes wrong
     */

    @Override
    public MdInlineCode handleContentChange(@NotNull MdInlineCode element, @NotNull TextRange range, String newContent) throws IncorrectOperationException {
        return handleContentChange(element, newContent);
    }

    @Override
    public MdInlineCode handleContentChange(@NotNull MdInlineCode element, String newContent) throws IncorrectOperationException {
        return (MdInlineCode) element.setContent(newContent);
    }

    @NotNull
    @Override
    public TextRange getRangeInElement(@NotNull MdInlineCode element) {
        return element.getContentRange(false);
    }
}


text:
const vx_enum VX_IMAGE_FORMAT = VX_IMAGE_ATTRIBUTE_FORMAT;
const vx_enum VX_IMAGE_WIDTH  = VX_IMAGE_ATTRIBUTE_WIDTH;
const vx_enum VX_IMAGE_HEIGHT = VX_IMAGE_ATTRIBUTE_HEIGHT;
const vx_enum VX_MEMORY_TYPE_HOST = VX_IMPORT_TYPE_HOST;
const vx_enum VX_MEMORY_TYPE_NONE = VX_IMPORT_TYPE_NONE;
const vx_enum VX_THRESHOLD_THRESHOLD_VALUE = VX_THRESHOLD_ATTRIBUTE_THRESHOLD_VALUE;
const vx_enum VX_THRESHOLD_THRESHOLD_LOWER = VX_THRESHOLD_ATTRIBUTE_THRESHOLD_LOWER;
const vx_enum VX_THRESHOLD_THRESHOLD_UPPER = VX_THRESHOLD_ATTRIBUTE_THRESHOLD_UPPER;
typedef uintptr_t vx_map_id;
#endif

enum UserMemoryMode
{
    COPY, USER_MEM
};

vx_image convertCvMatToVxImage(vx_context context, cv::Mat image, bool toCopy);
cv::Mat copyVxImageToCvMat(vx_image ovxImage);
void swapVxImage(vx_image ovxImage);
vx_status createProcessingGraph(vx_image inputImage, vx_image outputImage, vx_graph& graph);
int ovxDemo(std::string inputPath, UserMemoryMode mode);

vx_image convertCvMatToVxImage(vx_context context, cv::Mat image, bool toCopy)
{
    if (!(!image.empty() && image.dims <= 2 && image.channels() == 1))
        throw std::runtime_error("Invalid format");

    vx_uint32 width  = image.cols;
    vx_uint32 height = image.rows;

    vx_df_image color;
    switch (image.depth())
    {
    case CV_8U:
        color = VX_DF_IMAGE_U8;
        break;
    case CV_16U:
        color = VX_DF_IMAGE_U16;
        break;
    case CV_16S:
        color = VX_DF_IMAGE_S16;
        break;
    case CV_32S:
        color = VX_DF_IMAGE_S32;
        break;
    default:
        throw std::runtime_error("Invalid format");
        break;
    }

    vx_imagepatch_addressing_t addr;
    addr.dim_x = width;
    addr.dim_y = height;
    addr.stride_x = (vx_uint32)image.elemSize();
    addr.stride_y = (vx_uint32)image.step.p[0];
    vx_uint8* ovxData = image.data;

    vx_image ovxImage;
    if (toCopy)
    {
        ovxImage = vxCreateImage(context, width, height, color);
        if (vxGetStatus((vx_reference)ovxImage) != VX_SUCCESS)
            throw std::runtime_error("Failed to create image");
        vx_rectangle_t rect;

        vx_status status = vxGetValidRegionImage(ovxImage, &rect);
        if (status != VX_SUCCESS)
            throw std::runtime_error("Failed to get valid region");

#ifdef VX_VERSION_1_1
        status = vxCopyImagePatch(ovxImage, &rect, 0, &addr, ovxData, VX_WRITE_ONLY, VX_MEMORY_TYPE_HOST);
        if (status != VX_SUCCESS)
            throw std::runtime_error("Failed to copy image patch");
#else
        status = vxAccessImagePatch(ovxImage, &rect, 0, &addr, (void**)&ovxData, VX_WRITE_ONLY);
        if (status != VX_SUCCESS)
            throw std::runtime_error("Failed to access image patch");
        status = vxCommitImagePatch(ovxImage, &rect, 0, &addr, ovxData);
        if (status != VX_SUCCESS)
            throw std::runtime_error("Failed to commit image patch");
#endif
    }
    else
    {
        ovxImage = vxCreateImageFromHandle(context, color, &addr, (void**)&ovxData, VX_MEMORY_TYPE_HOST);
        if (vxGetStatus((vx_reference)ovxImage) != VX_SUCCESS)
            throw std::runtime_error("Failed to create image from handle");
    }

    return ovxImage;
}

cv::Mat copyVxImageToCvMat(vx_image ovxImage)
{
    vx_status status;
    vx_df_image df_image = 0;
    vx_uint32 width, height;
    status = vxQueryImage(ovxImage, VX_IMAGE_FORMAT, &df_image, sizeof(vx_df_image));
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to query image");
    status = vxQueryImage(ovxImage, VX_IMAGE_WIDTH, &width, sizeof(vx_uint32));
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to query image");
    status = vxQueryImage(ovxImage, VX_IMAGE_HEIGHT, &height, sizeof(vx_uint32));
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to query image");

    if (!(width > 0 && height > 0)) throw std::runtime_error("Invalid format");

    int depth;
    switch (df_image)
    {
    case VX_DF_IMAGE_U8:
        depth = CV_8U;
        break;
    case VX_DF_IMAGE_U16:
        depth = CV_16U;
        break;
    case VX_DF_IMAGE_S16:
        depth = CV_16S;
        break;
    case VX_DF_IMAGE_S32:
        depth = CV_32S;
        break;
    default:
        throw std::runtime_error("Invalid format");
        break;
    }

    cv::Mat image(height, width, CV_MAKE_TYPE(depth, 1));

    vx_rectangle_t rect;
    rect.start_x = rect.start_y = 0;
    rect.end_x = width; rect.end_y = height;

    vx_imagepatch_addressing_t addr;
    addr.dim_x = width;
    addr.dim_y = height;
    addr.stride_x = (vx_uint32)image.elemSize();
    addr.stride_y = (vx_uint32)image.step.p[0];
    vx_uint8* matData = image.data;

#ifdef VX_VERSION_1_1
    status = vxCopyImagePatch(ovxImage, &rect, 0, &addr, matData, VX_READ_ONLY, VX_MEMORY_TYPE_HOST);
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to copy image patch");
#else
    status = vxAccessImagePatch(ovxImage, &rect, 0, &addr, (void**)&matData, VX_READ_ONLY);
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to access image patch");
    status = vxCommitImagePatch(ovxImage, &rect, 0, &addr, matData);
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to commit image patch");
#endif

    return image;
}

void swapVxImage(vx_image ovxImage)
{
#ifdef VX_VERSION_1_1
    vx_status status;
    vx_memory_type_e memType;
    status = vxQueryImage(ovxImage, VX_IMAGE_MEMORY_TYPE, &memType, sizeof(vx_memory_type_e));
    if (status != VX_SUCCESS)
        throw std::runtime_error("Failed to query image");
    if (memType == VX_MEMORY_TYPE_NONE)
    {
        //was created by copying user data
        throw std::runtime_error("Image wasn't created from user handle");
    }
    else
    {
        //was created from user handle
        status = vxSwapImageHandle(ovxImage, NULL, NULL, 0);
        if (status != VX_SUCCESS)
            throw std::runtime_error("Failed to swap image handle");
    }
#else
    //not supported until OpenVX 1.1
    (void) ovxImage;
#endif
}

vx_status createProcessingGraph(vx_image inputImage, vx_image outputImage, vx_graph& graph)
{
    vx_status status;
    vx_context context = vxGetContext((vx_reference)inputImage);
    status = vxGetStatus((vx_reference)context);
    if(status != VX_SUCCESS) return status;

    graph = vxCreateGraph(context);
    status = vxGetStatus((vx_reference)graph);
    if (status != VX_SUCCESS) return status;

    vx_uint32 width, height;
    status = vxQueryImage(inputImage, VX_IMAGE_WIDTH, &width, sizeof(vx_uint32));
    if (status != VX_SUCCESS) return status;
    status = vxQueryImage(inputImage, VX_IMAGE_HEIGHT, &height, sizeof(vx_uint32));
    if (status != VX_SUCCESS) return status;

    // Intermediate images
    vx_image
        smoothed  = vxCreateVirtualImage(graph, 0, 0, VX_DF_IMAGE_VIRT),
        cannied   = vxCreateVirtualImage(graph, 0, 0, VX_DF_IMAGE_VIRT),
        halfImg   = vxCreateImage(context, width, height, VX_DF_IMAGE_U8),
        halfCanny = vxCreateImage(context, width, height, VX_DF_IMAGE_U8);

    vx_image virtualImages[] = {smoothed, cannied, halfImg, halfCanny};
    for(size_t i = 0; i < sizeof(virtualImages)/sizeof(vx_image); i++)
    {
        status = vxGetStatus((vx_reference)virtualImages[i]);
        if (status != VX_SUCCESS) return status;
    }

    // Constants
    vx_uint32 threshValue = 50;
    vx_threshold thresh = vxCreateThreshold(context, VX_THRESHOLD_TYPE_BINARY, VX_TYPE_UINT8);
    vxSetThresholdAttribute(thresh, VX_THRESHOLD_THRESHOLD_VALUE,
                            &threshValue, sizeof(threshValue));

    vx_uint32 threshCannyMin = 127;
    vx_uint32 threshCannyMax = 192;
    vx_threshold threshCanny = vxCreateThreshold(context, VX_THRESHOLD_TYPE_RANGE, VX_TYPE_UINT8);
    vxSetThresholdAttribute(threshCanny, VX_THRESHOLD_THRESHOLD_LOWER, &threshCannyMin,
                            sizeof(threshCannyMin));
    vxSetThresholdAttribute(threshCanny, VX_THRESHOLD_THRESHOLD_UPPER, &threshCannyMax,
                            sizeof(threshCannyMax));
    vx_float32 alphaValue = 0.5;
    vx_scalar alpha = vxCreateScalar(context, VX_TYPE_FLOAT32, &alphaValue);

    // Sequence of meaningless image operations
    vx_node nodes[] = {
        vxGaussian3x3Node(graph, inputImage, smoothed),
        vxCannyEdgeDetectorNode(graph, smoothed, threshCanny, 3, VX_NORM_L2, cannied),
        vxAccumulateWeightedImageNode(graph, inputImage, alpha, halfImg),
        vxAccumulateWeightedImageNode(graph, cannied, alpha, halfCanny),
        vxAddNode(graph, halfImg, halfCanny, VX_CONVERT_POLICY_SATURATE, outputImage)
    };

    for (size_t i = 0; i < sizeof(nodes) / sizeof(vx_node); i++)
    {
        status = vxGetStatus((vx_reference)nodes[i]);
        if (status != VX_SUCCESS) return status;
    }

    status = vxVerifyGraph(graph);
    return status;
}

int ovxDemo(std::string inputPath, UserMemoryMode mode)
{
    cv::Mat image = cv::imread(inputPath, cv::IMREAD_GRAYSCALE);
    if (image.empty()) return -1;

    //check image format
    if (image.depth() != CV_8U || image.channels() != 1) return -1;

    vx_status status;
    vx_context context = vxCreateContext();
    status = vxGetStatus((vx_reference)context);
    if (status != VX_SUCCESS) return status;

    //put user data from cv::Mat to vx_image
    vx_image ovxImage;
    ovxImage = convertCvMatToVxImage(context, image, mode == COPY);

    vx_uint32 width = image.cols, height = image.rows;

    vx_image ovxResult;
    cv::Mat output;
    if (mode == COPY)
    {
        //we will copy data from vx_image to cv::Mat
        ovxResult = vxCreateImage(context, width, height, VX_DF_IMAGE_U8);
        if (vxGetStatus((vx_reference)ovxResult) != VX_SUCCESS)
            throw std::runtime_error("Failed to create image");
    }
    else
    {
        //create vx_image based on user data, no copying required
        output = cv::Mat(height, width, CV_8U, cv::Scalar(0));
        ovxResult = convertCvMatToVxImage(context, output, false);
    }

    vx_graph graph;
    status = createProcessingGraph(ovxImage, ovxResult, graph);
    if (status != VX_SUCCESS) return status;

    // Graph execution
    status = vxProcessGraph(graph);
    if (status != VX_SUCCESS) return status;

    //getting resulting image in cv::Mat
    if (mode == COPY)
    {
        output = copyVxImageToCvMat(ovxResult);
    }
    else
    {
        //we should take user memory back from vx_image before using it (even before reading)
        swapVxImage(ovxResult);
    }

    //here output goes
    cv::imshow("processing result", output);
    cv::waitKey(0);

    //we need to take user memory back before releasing the image
    if (mode == USER_MEM)
        swapVxImage(ovxImage);

    cv::destroyAllWindows();

    status = vxReleaseContext(&context);
    return status;
}

int main(int argc, char *argv[])
{
    const std::string keys =
        "{help h usage ? | | }"
        "{image    | <none> | image to be processed}"
        "{mode | copy | user memory interaction mode: \n"
        "copy: create VX images and copy data to/from them\n"
        "user_mem: use handles to user-allocated memory}"
        ;

    cv::CommandLineParser parser(argc, argv, keys);
    parser.about("OpenVX interoperability sample demonstrating standard OpenVX API."
                 "The application loads an image, processes it with OpenVX graph and outputs result in a window");
    if (parser.has("help"))
    {
        parser.printMessage();
        return 0;
    }
    std::string imgPath = parser.get<std::string>("image");
    std::string modeString = parser.get<std::string>("mode");
    UserMemoryMode mode;
    if(modeString == "copy")
    {
        mode = COPY;
    }
    else if(modeString == "user_mem")
    {
        mode = USER_MEM;
    }
    else if(modeString == "map")
    {
        std::cerr << modeString << " is not implemented in this sample" << std::endl;
        return -1;
    }
    else
    {
        std::cerr << modeString << ": unknown memory mode" << std::endl;
        return -1;
    }

    if (!parser.check())
    {
        parser.printErrors();
        return -1;
    }

    return ovxDemo(imgPath, mode);
}


text:
package io.luna.game.model.mob.block;

import io.luna.game.model.mob.Graphic;
import io.luna.game.model.mob.Npc;
import io.luna.game.model.mob.Player;
import io.luna.game.model.mob.block.UpdateFlagSet.UpdateFlag;
import io.luna.net.codec.ByteMessage;
import io.luna.net.codec.ByteOrder;

/**
 * An {@link UpdateBlock} implementation for the {@code GRAPHIC} update block.
 *
 * @author lare96 <http://github.org/lare96>
 */
public final class GraphicUpdateBlock extends UpdateBlock {

    /**
     * Creates a new {@link GraphicUpdateBlock}.
     */
    public GraphicUpdateBlock() {
        super(UpdateFlag.GRAPHIC);
    }

    @Override
    public void encodeForPlayer(Player player, ByteMessage msg) {
        Graphic graphic = unwrap(player.getGraphic());
        msg.putShort(graphic.getId(), ByteOrder.LITTLE);
        msg.putInt(graphic.getHeight() << 16 | graphic.getDelay() & 0xFFFF);
    }

    @Override
    public void encodeForNpc(Npc npc, ByteMessage msg) {
        Graphic graphic = unwrap(npc.getGraphic());
        msg.putShort(graphic.getId());
        msg.putInt(graphic.getHeight() << 16 | graphic.getDelay() & 0xFFFF);
    }

    @Override
    public int getPlayerMask() {
        return 256;
    }

    @Override
    public int getNpcMask() {
        return 128;
    }
}


text:
import { ParserService } from '../ParserService';
export declare class JSONParserService extends ParserService {
    /**
     * Парсит JSON-строку и возвращает объект. В случае неуспеха возвращает undefined.
     * @param data
     * @returns {Object}
     */
    parseString(data: string): object;
    /**
     * Парсит текст, содержащийся в элементе, как JSON и возвращает объект или undefined.
     * @param element
     * @returns {undefined}
     */
    parseDOM(element: HTMLElement): object;
}