peers builtin; fixed stop request issuing in starter

musique/interpreter/builtin_functions.cc

@@ -1591,10 +1591,16 @@ static Result<Value> builtin_start(Interpreter &interpreter, std::span<Ast> args
 	{
 		return Try(interpreter.eval((Ast)ast));
 	});
+	// FIXME this should be in "finally" block
 	interpreter.starter.stop();
 	return result;
 }
 
+static Result<Value> builtin_peers(Interpreter &interpreter, std::vector<Value>)
+{
+	return Number(interpreter.starter.peers());
+}
+
 void Interpreter::register_builtin_functions()
 {
 	auto &global = *Env::global;
@@ -1624,6 +1630,7 @@ void Interpreter::register_builtin_functions()
 	global.force_define("oct",            builtin_oct);
 	global.force_define("par",            builtin_par);
 	global.force_define("partition",      builtin_partition);
+	global.force_define("peers",          builtin_peers);
 	global.force_define("permute",        builtin_permute);
 	global.force_define("pgmchange",      builtin_program_change);
 	global.force_define("pick",           builtin_pick);

musique/interpreter/starter.cc

@@ -23,6 +23,7 @@ struct State
 	ableton::Link *link;
 	std::mutex mutex;
 	std::condition_variable condition;
+	std::stop_source source;
 };
 
 void Starter::start()
@@ -46,24 +47,28 @@ void Starter::start()
 	{
 		if (is_playing) {
 			state->condition.notify_all();
+			state->source.request_stop();
 		}
 	});
 
-	std::jthread starter([quantum = quantum, state = state](std::stop_token token)
+	std::thread starter([quantum = quantum, state = state, token = state->source.get_token()]()
 	{
 		auto counter = 0;
 		while (!token.stop_requested()) {
 			auto const time = state->link->clock().micros();
 			auto sessionState = state->link->captureAppSessionState();
-			if (counter == 2) {
+			if (counter == 10) {
 				sessionState.setIsPlaying(true, time);
 				state->link->commitAppSessionState(sessionState);
+				state->condition.notify_all();
+				state->source.request_stop();
 				return;
 			}
 			auto const phase = sessionState.phaseAtTime(time, quantum);
 			counter += phase == 0;
 		}
 	});
+	starter.detach();
 
 	std::unique_lock lock(state->mutex);
 	state->condition.wait(lock);
@@ -79,3 +84,11 @@ void Starter::stop()
 	sessionState.setIsPlaying(false, time);
 	link.commitAppSessionState(sessionState);
 }
+
+size_t Starter::peers() const
+{
+	ensure(impl != nullptr, "Starter wasn't initialized properly");
+	auto &link = impl->link;
+
+	return link.numPeers();
+}

musique/interpreter/starter.hh

@@ -9,6 +9,8 @@ struct Starter
 	void start();
 	void stop();
 
+	size_t peers() const;
+
 	struct Implementation;
 	std::shared_ptr<Implementation> impl;
 };