qa: trying to understand pubsub_install tor aborts

CDEPS/tor/004.patch

@@ -0,0 +1,68 @@
+diff --git a/src/lib/pubsub/pubsub_check.c b/src/lib/pubsub/pubsub_check.c
+index 99e604d715..a5cc4b7658 100644
+--- a/src/lib/pubsub/pubsub_check.c
++++ b/src/lib/pubsub/pubsub_check.c
+@@ -25,6 +25,7 @@
+ #include "lib/malloc/malloc.h"
+ #include "lib/string/compat_string.h"
+
++#include <stdio.h>
+ #include <string.h>
+
+ static void pubsub_adjmap_add(pubsub_adjmap_t *map,
+@@ -343,21 +344,27 @@ lint_message(const pubsub_adjmap_t *map, message_id_t msg)
+     log_warn(LD_MESG|LD_BUG,
+              "Message \"%s\" has subscribers, but no publishers.",
+              get_message_id_name(msg));
++    fprintf(stderr, "SBSDEBUG: n_pub == 0 for %s\n", get_message_id_name(msg));
+     ok = false;
+   } else if (n_sub == 0) {
+     log_warn(LD_MESG|LD_BUG,
+              "Message \"%s\" has publishers, but no subscribers.",
+              get_message_id_name(msg));
++    fprintf(stderr, "SBSDEBUG: n_sub == 0 for %s\n", get_message_id_name(msg));
+     ok = false;
+   }
+
+   /* Check the message graph topology. */
+-  if (lint_message_graph(map, msg, pub, sub) < 0)
++  if (lint_message_graph(map, msg, pub, sub) < 0) {
++    fprintf(stderr, "SBSDEBUG: lint_message_graph failed for %s\n", get_message_id_name(msg));
+     ok = false;
++  }
+
+   /* Check whether the messages have the same fields set on them. */
+-  if (lint_message_consistency(msg, pub, sub) < 0)
++  if (lint_message_consistency(msg, pub, sub) < 0) {
++    fprintf(stderr, "SBSDEBUG: lint_message_consistency failed for %s\n", get_message_id_name(msg));
+     ok = false;
++  }
+
+   if (!ok) {
+     /* There was a problem -- let's log all the publishers and subscribers on
+@@ -385,6 +392,7 @@ pubsub_adjmap_check(const pubsub_adjmap_t *map)
+   bool all_ok = true;
+   for (unsigned i = 0; i < map->n_msgs; ++i) {
+     if (lint_message(map, i) < 0) {
++      fprintf(stderr, "SBSDEBUG: lint_message failed for %u %s\n", i, get_message_id_name((message_id_t)i));
+       all_ok = false;
+     }
+   }
+@@ -401,11 +409,15 @@ pubsub_builder_check(pubsub_builder_t *builder)
+   pubsub_adjmap_t *map = pubsub_build_adjacency_map(builder->items);
+   int rv = -1;
+
+-  if (!map)
++  if (!map) {
++    fprintf(stderr, "SBSDEBUG: pubsub_build_adjacency_map failed\n");
+     goto err; // should be impossible
++  }
+
+-  if (pubsub_adjmap_check(map) < 0)
++  if (pubsub_adjmap_check(map) < 0) {
++    fprintf(stderr, "SBSDEBUG: pubsub_adjmap_check failed\n");
+     goto err;
++  }
+
+   rv = 0;
+  err:

internal/cmd/buildtool/linuxcdeps.go

@@ -44,7 +44,9 @@ func linuxCdepsBuildMain(name string, deps buildtoolmodel.Dependencies) {
 		"-fPIC", // makes more sense than -fPIE given that we're building a library
 		"-fsanitize=bounds",
 		"-fsanitize-undefined-trap-on-error",
+		"-fsanitize=thread",
 		"-O2",
+		"-g",
 	}
 	destdir := runtimex.Try1(filepath.Abs(filepath.Join( // must be absolute
 		"internal", "libtor", "linux", runtime.GOARCH,

internal/cmd/testtorsf/main.go

@@ -0,0 +1,69 @@
+//go:build ooni_libtor
+
+package main
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/apex/log"
+	"github.com/ooni/probe-cli/v3/internal/experiment/vanillator"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
+	"github.com/ooni/probe-cli/v3/internal/model"
+	"github.com/ooni/probe-cli/v3/internal/model/mocks"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
+)
+
+func runit() {
+	measurer := vanillator.NewExperimentMeasurer(vanillator.Config{
+		//DisablePersistentDatadir: false,
+		DisableProgress: false,
+		//RendezvousMethod:         "",
+	})
+	meas := &model.Measurement{}
+	err := measurer.Run(
+		context.Background(),
+		&model.ExperimentArgs{
+			Callbacks:   model.NewPrinterCallbacks(model.DiscardLogger),
+			Measurement: meas,
+			Session: &mocks.Session{
+				MockDefaultHTTPClient: func() model.HTTPClient {
+					return http.DefaultClient
+				},
+				MockKeyValueStore: func() model.KeyValueStore {
+					return &kvstore.Memory{}
+				},
+				MockLogger: func() model.Logger {
+					return log.Log
+				},
+				MockSoftwareName: func() string {
+					return "miniooni"
+				},
+				MockSoftwareVersion: func() string {
+					return "0.1.0-dev"
+				},
+				MockTempDir: func() string {
+					return "x/tmp"
+				},
+				MockTunnelDir: func() string {
+					return "x/tunnel"
+				},
+				MockUserAgent: func() string {
+					return model.HTTPHeaderUserAgent
+				},
+			},
+		},
+	)
+	runtimex.PanicOnError(err, "measurer.Run failed")
+	tk := meas.TestKeys.(*vanillator.TestKeys)
+	runtimex.Assert(tk.Success, "did not succeed")
+}
+
+func main() {
+	for {
+		runit()
+		log.Info("************* now let's wait a bit ********************************")
+		time.Sleep(45 * time.Second)
+	}
+}

internal/libtor/enabled.go

@@ -57,6 +57,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"runtime"
 	"sync"
 
 	"github.com/cretz/bine/process"
@@ -149,6 +150,13 @@ var ErrNonzeroExitCode = errors.New("libtor: command completed with nonzero exit
 // runtor runs tor until completion and ensures that tor exits when
 // the given ctx is cancelled or its deadline expires.
 func (p *torProcess) runtor(ctx context.Context, cc net.Conn, args ...string) {
+	// make sure we lock to an OS thread otherwise the goroutine can get
+	// preempted midway and causes data races
+	//
+	// See https://github.com/ooni/probe/issues/2406#issuecomment-1479138677
+	runtime.LockOSThread()
+	defer runtime.UnlockOSThread()
+
 	// wait for Start or context to expire
 	select {
 	case <-p.awaitStart:

internal/tunnel/torembed.go

@@ -1,4 +1,4 @@
-//go:build ooni_libtor && android
+//go:build ooni_libtor && (android || linux)
 
 package tunnel
 

y/compile.bash

@@ -0,0 +1,3 @@
+#!/bin/bash
+set -euxo pipefail
+gcc -g -Wall -Wextra -fsanitize=thread -I internal/libtor/linux/amd64/include -L internal/libtor/linux/amd64/lib y/main.c -ltor -levent -lssl -lcrypto -lz -lm

y/main.c

@@ -0,0 +1,53 @@
+#include <tor_api.h>
+
+#include <pthread.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+
+static void *threadMain(void *ptr) {
+	int *fdp = (int*)ptr;
+	(void)sleep(45 /* second */);
+	(void)close(*fdp);
+	free(fdp);
+	return NULL;
+}
+
+int main() {
+	for (;;) {
+		tor_main_configuration_t *config = tor_main_configuration_new();
+		if (config == NULL) {
+			exit(1);
+		}
+		char *argv[] = {
+			"tor",
+			"Log",
+			"notice stderr",
+			"DataDirectory",
+			"./x",
+			NULL,
+		};
+		int argc = 5;
+		if (tor_main_configuration_set_command_line(config, argc, argv) != 0) {
+			exit(2);
+		}
+		int filedesc = tor_main_configuration_setup_control_socket(config);
+		if (filedesc < 0) {
+			exit(3);
+		}
+		int *fdp = malloc(sizeof(*fdp));
+		if (fdp == NULL) {
+			exit(4);
+		}
+		*fdp = filedesc;
+		pthread_t thread;
+		if (pthread_create(&thread, NULL, threadMain, /* move */ fdp) != 0) {
+			exit(5);
+		}
+		tor_run_main(config);
+		if (pthread_join(thread, NULL) != 0) {
+			exit(6);
+		}
+		fprintf(stderr, "********** doing another round\n");
+	}
+}