Inside Bitcoin

251            socket_pair = self.wakeup_socket_pair
252            if socket_pair is not None:
253                socket_pair[0].send("CloseWakeup".encode())
254                logger.debug("Waking up forwarding thread")
255        except OSError as e:
256            logger.warning(f"Error waking up forwarding thread: {e}")
257            pass
258
259
260# Wrapper for thread.join(), which may throw for daemon threads (in late stages of finalization).
261# Return True if the thread is no longer active (join succeeded), False otherwise
262# See PR #34863 for more details on using daemon threads.
263def try_join_daemon_thread(thread, timeout=0) -> bool:
264    try:
265        thread.join(timeout=timeout)
266        return not thread.is_alive()
267    except Exception as e:
268        logger.debug(f"Exception in thread.join, {e}")
269        return True
270
271class Socks5Server():
272    def __init__(self, conf):
273        self.conf = conf
274        self.s = socket.socket(conf.af)
275        self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
276        # When using dynamic port allocation (port=0), ensure we don't get a
277        # port that conflicts with the test framework's static port range.
278        if conf.addr[1] == 0:
279            set_ephemeral_port_range(self.s)
280        self.s.bind(conf.addr)
281        # When port=0, the OS assigns an available port. Update conf.addr
282        # to reflect the actual bound address so callers can use it.
283        self.conf.addr = self.s.getsockname()
284        self.s.listen(5)
285        # Set to False when stop is initiated
286        self._running = False
287        self._running_lock = threading.Lock()
288        self.thread = None
289        self.queue = queue.Queue() # report connections and exceptions to client
290        self.keep_alive = conf.keep_alive
291        # Store the background handlers, needed for clean shutdown
292        # Append-only array, completed handlers are set to None
293        self._handlers = []
294        self._handlers_lock = threading.Lock()
295
296    def is_running(self) -> bool:
297        with self._running_lock:
298            return self._running
299
300    def set_running(self, new_value: bool):
301        with self._running_lock:
302            self._running = new_value
303
304    def run(self):
305        while self.is_running():
306            (sockconn, _) = self.s.accept()
307            if self.is_running():
308                conn = Socks5Connection(self, sockconn)
309                # Use "daemon" threads, see PR #34863 for more discussion.
310                thread = threading.Thread(None, conn.handle, daemon=True)
311                with self._handlers_lock:
312                    conn.handler_index = len(self._handlers)
313                    self._handlers.append((thread, conn))
314                    assert(conn.handler_index < len(self._handlers))
315                thread.start()
316
317    def remove_handler(self, handler_index):
318        with self._handlers_lock:
319            if handler_index < len(self._handlers):
320                if self._handlers[handler_index] is not None:
321                    self._handlers[handler_index] = None
322                    logger.debug(f"Handler {handler_index} removed")
323
324    def start(self):
325        assert not self.is_running()
326        self.set_running(True)
327        self.thread = threading.Thread(None, self.run, daemon=True)
328        self.thread.start()
329
330    def stop(self):
331        self.set_running(False)
332        # connect to self to end run loop
333        s = socket.socket(self.conf.af)
334        s.connect(self.conf.addr)
335        s.close()
336        self.thread.join()
337        # if there are active handlers, close them
338        with self._handlers_lock:
339            items = list(self._handlers)
340        for i, item in enumerate(items):
341            if item is None:
342                continue
343            thread, conn = item
344            # check if thread is still active
345            if not try_join_daemon_thread(thread, timeout=0):
346                conn.wakeup()
347                if try_join_daemon_thread(thread, timeout=2):
348                    logger.debug(f"Stop(): Handler {i} thread joined")
349                else:
350                    logger.warning(f"Stop(): Handler thread {i} didn't finish after force close")
351
352def start_socks5_server(destinations_factory):
353    config = Socks5Configuration()
354    config.addr = ("127.0.0.1", 0) # Use port=0 to let the OS pick one. The actual port is later in server.conf.addr[1].
355    config.unauth = True
356    config.auth = True
357    config.destinations_factory = destinations_factory
358
359    server = Socks5Server(config)
360    server.start()
361
362    return server
363