Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
73dde99e3a | ||
|
|
ecc8944006 | ||
|
|
dc15e2d04b | ||
|
|
ff1fc25d77 |
@@ -38,6 +38,14 @@
|
|||||||
## Plan (предыдущая задача — выполнена)
|
## Plan (предыдущая задача — выполнена)
|
||||||
1–7. Docker E2E пайплайна «install как человек → serve → визард → воркер → джоб» — выполнено, см. Progress ниже.
|
1–7. Docker E2E пайплайна «install как человек → serve → визард → воркер → джоб» — выполнено, см. Progress ниже.
|
||||||
|
|
||||||
|
## Progress (день: конкурентность в агенте — параллелизм через нашу архитектуру)
|
||||||
|
- ✅ По просьбе «реализовать параллелизм через нашу архитектуру»: `WORKER_CONCURRENCY` (поле `concurrency` в конфиге визарда, шаг Machine) — агент регистрируется один раз и ведёт N циклов claim→выполнение→upload под одним worker id. N шардов обрабатываются параллельно на одной машине, используя таски координатора как единицу параллелизма; SDK не менялся.
|
||||||
|
- ✅ Реализация: `Config.Concurrency` + env; `Daemon.RunForever` → register once → N goroutine-циклов (общий счётчик MaxTasks под мьютексом); визард: поле «Concurrent task loops» + конфиг; тест с маркер-скриптом доказывает 3 параллельных исполнения (race-тесты зелёные после мьютекса в fake).
|
||||||
|
- ✅ Измерено на релизном коде в Docker (один воркер, 8 шардов): concurrency=1 → 12s; concurrency=4 → 4s (**3×**). Плюс прежний `similarity-search-parallel` (потоки внутри шарда) композируется с конкурентностью.
|
||||||
|
- ✅ Релиз v1.1.0-alpha.20 (бинарники + wheel); полный гейт: race + lint 0 issues + pytest 213.
|
||||||
|
- ✅ Бинарник worker-agent на машине пользователя обновлён до alpha.20.
|
||||||
|
- GIL-высвобождения в RDKit нет (проверено: `RDKIT_ALLOW_GIL_RELEASE` не помогает), поэтому внутришардовые потоки не ускоряют чистый RDKit-путь — конкурентность задач это и компенсирует.
|
||||||
|
|
||||||
## Progress (утро: similarity-search-parallel)
|
## Progress (утро: similarity-search-parallel)
|
||||||
- ✅ Новый workload `similarity-search-parallel@1.0.0` (отдельная версия, как просил пользователь): подкласс `SimilaritySearchSDKWorkload` + параллельное ядро `search_parallel/core.py` — fingerprinting+скоринг шарда через `ThreadPoolExecutor` (параметр `threads`, default CPU count); `pool.map` сохраняет порядок строк, поэтому merge идентичен последовательному (`_HeapEntry`) и результат **байт-в-байт** равен `similarity-search` при любом числе потоков.
|
- ✅ Новый workload `similarity-search-parallel@1.0.0` (отдельная версия, как просил пользователь): подкласс `SimilaritySearchSDKWorkload` + параллельное ядро `search_parallel/core.py` — fingerprinting+скоринг шарда через `ThreadPoolExecutor` (параметр `threads`, default CPU count); `pool.map` сохраняет порядок строк, поэтому merge идентичен последовательному (`_HeapEntry`) и результат **байт-в-байт** равен `similarity-search` при любом числе потоков.
|
||||||
- ✅ Тесты: байт-в-байт vs эталон для threads 1/2/4 с намеренными связями (изомеры, дубликаты), executor-прогон, валидация параметров, регистрация манифеста; 213 pytest зелёные.
|
- ✅ Тесты: байт-в-байт vs эталон для threads 1/2/4 с намеренными связями (изомеры, дубликаты), executor-прогон, валидация параметров, регистрация манифеста; 213 pytest зелёные.
|
||||||
|
|||||||
@@ -45,7 +45,7 @@ func runAgent(args []string) error {
|
|||||||
return fmt.Errorf("--coordinator-url, --token, and --work-dir are required")
|
return fmt.Errorf("--coordinator-url, --token, and --work-dir are required")
|
||||||
}
|
}
|
||||||
if *taskRunner == "" {
|
if *taskRunner == "" {
|
||||||
*taskRunner = "python -m scimesh.worker.task"
|
*taskRunner = "python -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||||
|
|||||||
@@ -236,9 +236,9 @@ func stopAgents(agents []*exec.Cmd) {
|
|||||||
// system `python`.
|
// system `python`.
|
||||||
func defaultTaskRunner(venvPython string) string {
|
func defaultTaskRunner(venvPython string) string {
|
||||||
if runtimeStatus(venvPython) {
|
if runtimeStatus(venvPython) {
|
||||||
return venvPython + " -m scimesh.worker.task"
|
return venvPython + " -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
return "python -m scimesh.worker.task"
|
return "python -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
|
|
||||||
// ensureRuntime creates the managed venv and installs scimesh into it, unless
|
// ensureRuntime creates the managed venv and installs scimesh into it, unless
|
||||||
|
|||||||
@@ -88,8 +88,12 @@ func CheckEnvironment(ctx context.Context) CheckReport {
|
|||||||
// execute with.
|
// execute with.
|
||||||
func CheckEnvironmentWithPython(ctx context.Context, python string) CheckReport {
|
func CheckEnvironmentWithPython(ctx context.Context, python string) CheckReport {
|
||||||
report := CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: true, Detail: python}}
|
report := CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: true, Detail: python}}
|
||||||
|
// The version comes from importlib.metadata, so the wizard can compare the
|
||||||
|
// installed package with the binary version and offer an upgrade. -I keeps
|
||||||
|
// the working directory out of sys.path, so a scimesh checkout in the
|
||||||
|
// wizard's cwd can never shadow the venv installation.
|
||||||
//nolint:gosec // G204: python is a resolved interpreter path, the argument list is constant
|
//nolint:gosec // G204: python is a resolved interpreter path, the argument list is constant
|
||||||
cmd := exec.CommandContext(ctx, python, "-c", "import scimesh; print(scimesh.__version__ if hasattr(scimesh, '__version__') else 'installed')")
|
cmd := exec.CommandContext(ctx, python, "-I", "-c", "import importlib.metadata as m; print(m.version('scimesh'))")
|
||||||
out, err := cmd.Output()
|
out, err := cmd.Output()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// The worker executes workloads by spawning scimesh's task runner, so
|
// The worker executes workloads by spawning scimesh's task runner, so
|
||||||
|
|||||||
@@ -112,7 +112,7 @@ func LoadConfig() (*Config, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(runner) == 0 {
|
if len(runner) == 0 {
|
||||||
runner = []string{"python", "-m", "scimesh.worker.task"}
|
runner = []string{"python", "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
maxTasks := 0
|
maxTasks := 0
|
||||||
if raw := os.Getenv("MAX_TASKS"); raw != "" {
|
if raw := os.Getenv("MAX_TASKS"); raw != "" {
|
||||||
|
|||||||
@@ -112,7 +112,7 @@ func (f *ConfigFile) Config() (*Config, error) {
|
|||||||
config.TaskRunner = f.TaskRunner
|
config.TaskRunner = f.TaskRunner
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) == 0 {
|
if len(config.TaskRunner) == 0 {
|
||||||
config.TaskRunner = []string{"python", "-m", "scimesh.worker.task"}
|
config.TaskRunner = []string{"python", "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
config.PollInterval = 2 * time.Second
|
config.PollInterval = 2 * time.Second
|
||||||
config.RequestTimeout = 30 * time.Second
|
config.RequestTimeout = 30 * time.Second
|
||||||
|
|||||||
@@ -323,7 +323,7 @@ func (s *Server) ensureVenvTaskRunner() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if venv := s.venvPython(); venv != "" {
|
if venv := s.venvPython(); venv != "" {
|
||||||
file.TaskRunner = []string{venv, "-m", "scimesh.worker.task"}
|
file.TaskRunner = []string{venv, "-I", "-m", "scimesh.worker.task"}
|
||||||
if payload, err := json.MarshalIndent(file, "", " "); err == nil {
|
if payload, err := json.MarshalIndent(file, "", " "); err == nil {
|
||||||
_ = os.WriteFile(s.cfgPath, append(payload, '\n'), 0o600)
|
_ = os.WriteFile(s.cfgPath, append(payload, '\n'), 0o600)
|
||||||
}
|
}
|
||||||
@@ -428,7 +428,7 @@ func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) {
|
|||||||
// workloads execute through scimesh's task runner, which lives in the venv.
|
// workloads execute through scimesh's task runner, which lives in the venv.
|
||||||
if len(file.TaskRunner) == 0 {
|
if len(file.TaskRunner) == 0 {
|
||||||
if venv := s.venvPython(); venv != "" {
|
if venv := s.venvPython(); venv != "" {
|
||||||
file.TaskRunner = []string{venv, "-m", "scimesh.worker.task"}
|
file.TaskRunner = []string{venv, "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := agent.SaveConfigFile(s.cfgPath, file); err != nil {
|
if err := agent.SaveConfigFile(s.cfgPath, file); err != nil {
|
||||||
@@ -454,6 +454,9 @@ func (s *Server) handleTest(w http.ResponseWriter, r *http.Request) {
|
|||||||
// checking the bare system python3 would keep reporting scimesh as
|
// checking the bare system python3 would keep reporting scimesh as
|
||||||
// missing even though the worker would run with the venv.
|
// missing even though the worker would run with the venv.
|
||||||
report := agent.RunCheck(r.Context(), url, s.venvPython(), req.Token, req.WorkerKey, req.UserserviceURL)
|
report := agent.RunCheck(r.Context(), url, s.venvPython(), req.Token, req.WorkerKey, req.UserserviceURL)
|
||||||
|
if report.Scimesh.OK {
|
||||||
|
report.Scimesh = ensureMatchingScimeshVersion(report.Scimesh)
|
||||||
|
}
|
||||||
writeJSON(w, http.StatusOK, report)
|
writeJSON(w, http.StatusOK, report)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -628,3 +631,26 @@ func truncate(s string, n int) string {
|
|||||||
}
|
}
|
||||||
return s[:n] + "…"
|
return s[:n] + "…"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ensureMatchingScimeshVersion flips a green scimesh check to a stale one when
|
||||||
|
// the installed package does not match the worker-agent's own version: a
|
||||||
|
// version-locked wheel is the only supported runtime, and a mismatch means the
|
||||||
|
// workload catalog the worker advertises is not what it executes. The wizard
|
||||||
|
// UI then offers the Install button again. Dev builds have no release wheel,
|
||||||
|
// so they skip the comparison.
|
||||||
|
func ensureMatchingScimeshVersion(item agent.CheckItem) agent.CheckItem {
|
||||||
|
if agent.Version == "" || agent.Version == "dev" {
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
want := agent.NormalizePEP440(agent.Version)
|
||||||
|
got := strings.TrimSpace(item.Detail)
|
||||||
|
if got == "" || got == want {
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
item.OK = false
|
||||||
|
item.Detail = fmt.Sprintf(
|
||||||
|
"installed scimesh %s, but this worker-agent (%s) needs %s — press Install to upgrade",
|
||||||
|
got, agent.Version, want,
|
||||||
|
)
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
|||||||
@@ -429,8 +429,8 @@ func TestStartPinsTheVenvTaskRunner(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) != 3 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-m" || config.TaskRunner[2] != "scimesh.worker.task" {
|
if len(config.TaskRunner) != 4 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-I" || config.TaskRunner[2] != "-m" || config.TaskRunner[3] != "scimesh.worker.task" {
|
||||||
t.Errorf("task runner = %v, want the venv python runner", config.TaskRunner)
|
t.Errorf("task runner = %v, want the venv python runner with -I", config.TaskRunner)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -450,8 +450,8 @@ func TestSaveConfigPinsVenvRunnerWhenPresent(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) != 3 || config.TaskRunner[0] != venvPython {
|
if len(config.TaskRunner) != 4 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-I" {
|
||||||
t.Errorf("task runner = %v, want the venv python", config.TaskRunner)
|
t.Errorf("task runner = %v, want the venv python with -I", config.TaskRunner)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -462,7 +462,7 @@ func TestTestProbesTheVenvPythonAfterInstall(t *testing.T) {
|
|||||||
// a fake scimesh version so the preflight goes green through the venv.
|
// a fake scimesh version so the preflight goes green through the venv.
|
||||||
venvPython := filepath.Join(server.dir, "venv", "bin", "python")
|
venvPython := filepath.Join(server.dir, "venv", "bin", "python")
|
||||||
_ = os.MkdirAll(filepath.Dir(venvPython), 0o755)
|
_ = os.MkdirAll(filepath.Dir(venvPython), 0o755)
|
||||||
_ = os.WriteFile(venvPython, []byte("#!/bin/sh\nif [ \"$1\" = \"-c\" ]; then echo 9.9.9-test; exit 0; fi\nexit 0\n"), 0o755)
|
_ = os.WriteFile(venvPython, []byte("#!/bin/sh\nfor a in \"$@\"; do if [ \"$a\" = \"-c\" ]; then echo 9.9.9-test; exit 0; fi; done\nexit 0\n"), 0o755)
|
||||||
|
|
||||||
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, base+"/api/test", strings.NewReader(`{"coordinator_url":"http://127.0.0.1:1"}`))
|
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, base+"/api/test", strings.NewReader(`{"coordinator_url":"http://127.0.0.1:1"}`))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
@@ -497,3 +497,26 @@ time=6 level=WARN msg="agent cycle failed" error="boom"
|
|||||||
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
|
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func testCheckScimeshVersion(t *testing.T, installed, binary string, wantOK bool, wantDetail string) {
|
||||||
|
t.Helper()
|
||||||
|
old := agent.Version
|
||||||
|
agent.Version = binary
|
||||||
|
t.Cleanup(func() { agent.Version = old })
|
||||||
|
item := ensureMatchingScimeshVersion(agent.CheckItem{Name: "scimesh", OK: true, Detail: installed})
|
||||||
|
if item.OK != wantOK {
|
||||||
|
t.Errorf("installed=%s binary=%s: ok=%v, want %v (%s)", installed, binary, item.OK, wantOK, item.Detail)
|
||||||
|
}
|
||||||
|
if wantDetail != "" && !strings.Contains(item.Detail, wantDetail) {
|
||||||
|
t.Errorf("detail = %q, want it to contain %q", item.Detail, wantDetail)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsureMatchingScimeshVersion(t *testing.T) {
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a20", "1.1.0-alpha.20", true, "")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a17", "1.1.0-alpha.20", false, "press Install to upgrade")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a16.dev7+gea0fb8c59.d20260803", "1.1.0-alpha.20", false, "needs 1.1.0a20")
|
||||||
|
// Dev builds and unknown versions never block.
|
||||||
|
testCheckScimeshVersion(t, "anything", "dev", true, "")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a20", "", true, "")
|
||||||
|
}
|
||||||
|
|||||||
@@ -232,7 +232,7 @@ function draftConfig(){
|
|||||||
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0,
|
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0,
|
||||||
concurrency:parseInt($('in-conc').value||'1',10)
|
concurrency:parseInt($('in-conc').value||'1',10)
|
||||||
};
|
};
|
||||||
if(state.venvPython)cfg.task_runner=[state.venvPython,'-m','scimesh.worker.task'];
|
if(state.venvPython)cfg.task_runner=[state.venvPython,'-I','-m','scimesh.worker.task'];
|
||||||
return cfg;
|
return cfg;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user