9c831d1 · opencog-atomspace-cog

Changed files (3) +30 -11

opencog/persist/cog-simple/CogSimpleIO.cc +12 -0

…	…	@@ -94,6 +94,9 @@ void CogSimpleStorage::storeAtom(const Handle& h, bool synchronous)
94	94
95	95	std::lock_guard<std::mutex> lck(_mtx);
96	96	do_send(msg);
	97
	98	// Flush the response.
	99	do_recv();
97	100	}
98	101
99	102	void CogSimpleStorage::storeValue(const Handle& h, const Handle& key)
…	…	@@ -111,6 +114,9 @@ void CogSimpleStorage::storeValue(const Handle& h, const Handle& key)
111	114
112	115	std::lock_guard<std::mutex> lck(_mtx);
113	116	do_send(msg);
	117
	118	// Flush the response.
	119	do_recv();
114	120	}
115	121
116	122	void CogSimpleStorage::updateValue(const Handle& h, const Handle& key,
…	…	@@ -129,6 +135,9 @@ void CogSimpleStorage::updateValue(const Handle& h, const Handle& key,
129	135
130	136	std::lock_guard<std::mutex> lck(_mtx);
131	137	do_send(msg);
	138
	139	// Flush the response.
	140	do_recv();
132	141	}
133	142
134	143	void CogSimpleStorage::loadValue(const Handle& h, const Handle& key)
…	…	@@ -370,6 +379,9 @@ std::string CogSimpleStorage::writeFrame(const Handle& hasp)
370	379	{
371	380	std::lock_guard<std::mutex> lck(_mtx);
372	381	do_send(msg);
	382
	383	// Flush the response.
	384	do_recv();
373	385	}
374	386
375	387	// Store short-forms

opencog/persist/cog-storage/CogChannel.cc +12 -8

…	…	@@ -385,18 +385,22 @@ void CogChannel<Client, Data>::reply_handler(const Msg& msg)
385	385	template<typename Client, typename Data>
386	386	void CogChannel<Client, Data>::barrier()
387	387	{
388		// Generate a unique barrier ID
	388	// Drain work queues.
	389	_msg_buffer.barrier();
	390
	391	// Server can complete barrier only after it receives it on
	392	// all of the open sockets; the random string is the uuid to
	393	// disambiguate this barrier from any others that might get
	394	// issued. The barrier is retired after all N of them are received.
389	395	static thread_local std::minstd_rand rng(std::random_device{}());
390	396	uint64_t rnd = (uint64_t(rng()) << 32) \| rng();
391	397
392		// Build the barrier message. Each worker thread will send this,
393		// opening its socket if needed. Server completes when all N arrive.
	398	std::lock_guard<std::mutex> lck(_sock_set_mtx);
394	399	char msg[64];
395		snprintf(msg, sizeof(msg), "(cog-barrier %u \"%016lx\")\n", NTHREADS, rnd);
396
397		Data dummy = Data();
398		Msg block{nullptr, nullptr, true, msg, dummy};
399		_msg_buffer.barrier(block);
	400	snprintf(msg, sizeof(msg), "(cog-barrier %zu \"%016lx\")\n",
	401	_open_socks.size(), rnd);
	402	for (int sockfd : _open_socks)
	403	send(sockfd, msg, strlen(msg), MSG_NOSIGNAL);
400	404	}
401	405
402	406	template<typename Client, typename Data>

opencog/persist/cog-storage/CogIO.cc +6 -3

…	…	@@ -91,7 +91,8 @@ void CogStorage::storeAtom(const Handle& h, bool synchronous)
91	91	msg = "(cog-set-value! " + Sexpr::encode_atom(h) +
92	92	"(Predicate \"-TruthValueKey-\") #f)\n";
93	93
94		_io_queue.enqueue_noreply(msg);
	94	Pkt pkt;
	95	_io_queue.enqueue(this, msg, pkt, &CogStorage::noop_const);
95	96	}
96	97
97	98	void CogStorage::removeAtom(AtomSpace* frame, const Handle& h, bool recursive)
…	…	@@ -115,7 +116,8 @@ void CogStorage::storeValue(const Handle& h, const Handle& key)
115	116	Sexpr::encode_atom(key) +
116	117	Sexpr::encode_value(h->getValue(key)) + ")\n";
117	118
118		_io_queue.enqueue_noreply(msg);
	119	Pkt pkt;
	120	_io_queue.enqueue(this, msg, pkt, &CogStorage::noop_const);
119	121	}
120	122
121	123	void CogStorage::updateValue(const Handle& h, const Handle& key,
…	…	@@ -127,7 +129,8 @@ void CogStorage::updateValue(const Handle& h, const Handle& key,
127	129	Sexpr::encode_atom(key) +
128	130	Sexpr::encode_value(delta) + ")\n";
129	131
130		_io_queue.enqueue_noreply(msg);
	132	Pkt pkt;
	133	_io_queue.enqueue(this, msg, pkt, &CogStorage::noop_const);
131	134	}
132	135
133	136	void CogStorage::loadValue(const Handle& h, const Handle& key)