<commit_before>/*
 * Engine.h
 *
 *  Created on: Jan 7, 2016
 *      Author: mad
 */

#ifndef INCLUDE_PHY_ENGINE_H_
#define INCLUDE_PHY_ENGINE_H_

#include <mutex>
#include <condition_variable>

#include <vnl/Basic.h>
#include <vnl/Pipe.h>
#include <vnl/Address.h>
#include <vnl/Message.h>
#include <vnl/Memory.h>
#include <vnl/Random.h>
#include <vnl/Queue.h>
#include <vnl/List.h>
#include <vnl/Util.h>


namespace vnl {

class Stream;
class Object;
class ThreadEngine;
class FiberEngine;

// running module in new thread
Address spawn(Object* object, Pipe* pipe = 0);


class Engine : public Basic {
public:
	Engine() {
		std::lock_guard<std::mutex> lock(static_mutex);
		instances->push_back(this);
	}
	
	virtual ~Engine() {
		assert(queue.empty());
		std::lock_guard<std::mutex> lock(static_mutex);
		instances->remove(this);
	}
	
	// thread safe
	virtual void receive(Message* msg) {
		assert(msg->dst || msg->isack);
		msg->gate = this;
		msg->rcv_time = vnl::currentTimeMicros();
		mutex.lock();
		queue.push(msg);
		cond.notify_all();
		if(!msg->isack) {
			num_received++;
		}
		for(auto& func : rcv_hooks) {
			func(msg);
		}
		mutex.unlock();<commit_msg>fix for wrong send_latency_sum<commit_after>/*
 * Engine.h
 *
 *  Created on: Jan 7, 2016
 *      Author: mad
 */

#ifndef INCLUDE_PHY_ENGINE_H_
#define INCLUDE_PHY_ENGINE_H_

#include <mutex>
#include <condition_variable>

#include <vnl/Basic.h>
#include <vnl/Pipe.h>
#include <vnl/Address.h>
#include <vnl/Message.h>
#include <vnl/Memory.h>
#include <vnl/Random.h>
#include <vnl/Queue.h>
#include <vnl/List.h>
#include <vnl/Util.h>


namespace vnl {

class Stream;
class Object;
class ThreadEngine;
class FiberEngine;

// running module in new thread
Address spawn(Object* object, Pipe* pipe = 0);


class Engine : public Basic {
public:
	Engine() {
		std::lock_guard<std::mutex> lock(static_mutex);
		instances->push_back(this);
	}
	
	virtual ~Engine() {
		assert(queue.empty());
		std::lock_guard<std::mutex> lock(static_mutex);
		instances->remove(this);
	}
	
	// thread safe
	virtual void receive(Message* msg) {
		assert(msg->dst || msg->isack);
		msg->gate = this;
		mutex.lock();
		queue.push(msg);
		cond.notify_all();
		if(!msg->isack) {
			msg->rcv_time = vnl::currentTimeMicros();
			num_received++;
		}
		for(auto& func : rcv_hooks) {
			func(msg);
		}
		mutex.unlock();
{'fraction_non_alphanumeric': 0.16170212765957448, 'fraction_numerical': 0.005106382978723404, 'mean_word_length': 4.259507829977629, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5245954', 'n_tokens_mistral': 970, 'n_tokens_neox': 894, 'n_words': 188}