MCPcopy Create free account
hub / github.com/antirez/ds4 / dist_run_coordinator

Function dist_run_coordinator

ds4_distributed.c:5695–5752  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

5693 ds4_tokens_free(&transcript);
5694 return 1;
5695 }
5696 d->plan_ready = true;
5697 ds4_tokens_free(&transcript);
5698 rc = 0;
5699 }
5700 return rc;
5701}
5702
5703/* =========================================================================
5704 * Standalone Coordinator Entrypoint
5705 * ========================================================================= */
5706
5707static int dist_run_coordinator(ds4_engine *engine, const ds4_dist_options *opt, const ds4_dist_generation_options *gen) {
5708 char err[256];
5709 int listen_fd = dist_open_listener(opt->listen_host, opt->listen_port, err, sizeof(err));
5710 if (listen_fd < 0) {
5711 fprintf(stderr, "ds4: distributed coordinator: %s\n", err);
5712 return 1;
5713 }
5714
5715 ds4_dist_coordinator_state state;
5716 memset(&state, 0, sizeof(state));
5717 state.engine = engine;
5718 state.model_id = (uint32_t)ds4_engine_model_id(engine);
5719 state.n_layers = (uint32_t)ds4_engine_layer_count(engine);
5720 state.local_start = opt->layers.start;
5721 state.local_end = dist_resolved_layer_end(opt, state.n_layers);
5722 state.ctx_size = gen && gen->ctx_size > 0 ? (uint32_t)gen->ctx_size : 0u;
5723 state.local_has_output = opt->layers.has_output;
5724 state.local_can_output_head = ds4_engine_has_output_head(engine);
5725 state.replay_check = opt->replay_check;
5726 state.debug = opt->debug;
5727 state.use_control_for_work = gen && gen->prompt;
5728 state.prefill_chunk = opt->prefill_chunk;
5729 state.prefill_window = opt->prefill_window;
5730 state.activation_bits = dist_activation_bits_or_default(opt->activation_bits);
5731 pthread_mutex_init(&state.mu, NULL);
5732
5733 char local_end[32];
5734 if (opt->layers.has_output) snprintf(local_end, sizeof(local_end), "output");
5735 else snprintf(local_end, sizeof(local_end), "%u", opt->layers.end);
5736 DIST_COORD_DEBUG(&state,
5737 "ds4: distributed coordinator: listening on %s:%d model_id=%u layers=%u local=%u:%s activation_bits=%u\n",
5738 opt->listen_host,
5739 opt->listen_port,
5740 state.model_id,
5741 state.n_layers,
5742 opt->layers.start,
5743 local_end,
5744 state.activation_bits);
5745
5746 ds4_dist_accept_ctx accept_ctx = {
5747 .state = &state,
5748 .listen_fd = listen_fd,
5749 };
5750 if (!gen || !gen->prompt) {
5751 dist_coordinator_accept_main(&accept_ctx);
5752 return 0;

Callers 1

ds4_dist_runFunction · 0.85

Calls 8

dist_open_listenerFunction · 0.85
ds4_engine_model_idFunction · 0.85
ds4_engine_layer_countFunction · 0.85
dist_resolved_layer_endFunction · 0.85

Tested by

no test coverage detected