Implement permanently running worker thread

use_develop doesn't work for some reason. Other envs will lack the
bitmsghash*.so and more code in the proofofwork will be covered.

.buildbot/ubuntu/Dockerfile

@@ -1,5 +1,7 @@
 FROM ubuntu:focal
 
+ENV DEBIAN_FRONTEND=noninteractive
+
 RUN apt-get update
 
 RUN apt-get install -yq software-properties-common
@@ -8,6 +10,7 @@ RUN apt-add-repository ppa:purplei2p/i2pd
 RUN apt-get update
 
 RUN apt-get install -yq --no-install-suggests --no-install-recommends \
+    build-essential libcap-dev libffi-dev libssl-dev \
     python3-dev python3-pip python3.9 python3.9-dev python3.9-venv sudo i2pd
 
 RUN echo 'builder ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers

minode/bitmsghash/bitmsghash.cpp

@@ -0,0 +1,203 @@
+// bitmessage cracker, build with g++ or MSVS to a shared library,
+// use via minode.proofofwork module
+#ifdef _WIN32
+#include "winsock.h"
+#include "windows.h"
+#define uint64_t unsigned __int64
+#else
+#include <arpa/inet.h>
+#include <pthread.h>
+#include <stdint.h>
+#endif
+#include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+#if defined(__APPLE__) || defined(__FreeBSD__) || defined (__DragonFly__) || defined (__OpenBSD__) || defined (__NetBSD__)
+#include <sys/types.h>
+#include <sys/sysctl.h>
+#endif
+
+#include "openssl/sha.h"
+
+#define PY_SSIZE_T_CLEAN
+#include <Python.h>
+
+#define HASH_SIZE 64
+#define BUFLEN 16384
+
+#if defined(__GNUC__)
+  #define EXPORT __attribute__ ((__visibility__("default")))
+#elif defined(_WIN32)
+  #define EXPORT __declspec(dllexport)
+#endif
+
+#ifndef __APPLE__
+#define ntohll(x) ( ( (uint64_t)(ntohl( (unsigned int)((x << 32) >> 32) )) << 32) | ntohl( ((unsigned int)(x >> 32)) ) )
+#endif
+
+unsigned long long max_val;
+unsigned char *initialHash;
+unsigned long long successval = 0;
+unsigned int numthreads = 0;
+
+#ifdef _WIN32
+DWORD WINAPI threadfunc(LPVOID param) {
+#else
+void * threadfunc(void* param) {
+#endif
+	unsigned int incamt = *((unsigned int*)param);
+	SHA512_CTX sha;
+	unsigned char buf[HASH_SIZE + sizeof(uint64_t)] = { 0 };
+	unsigned char output[HASH_SIZE] = { 0 };
+
+	memcpy(buf + sizeof(uint64_t), initialHash, HASH_SIZE);
+
+	unsigned long long tmpnonce = incamt;
+	unsigned long long * nonce = (unsigned long long *)buf;
+	unsigned long long * hash = (unsigned long long *)output;
+	while (successval == 0) {
+		tmpnonce += numthreads;
+
+		(*nonce) = ntohll(tmpnonce); /* increment nonce */
+		SHA512_Init(&sha);
+		SHA512_Update(&sha, buf, HASH_SIZE + sizeof(uint64_t));
+		SHA512_Final(output, &sha);
+		SHA512_Init(&sha);
+		SHA512_Update(&sha, output, HASH_SIZE);
+		SHA512_Final(output, &sha);
+
+		if (ntohll(*hash) < max_val) {
+			successval = tmpnonce;
+		}
+	}
+#ifdef _WIN32
+	return 0;
+#else
+	return NULL;
+#endif
+}
+
+void getnumthreads()
+{
+#ifdef _WIN32
+	DWORD_PTR dwProcessAffinity, dwSystemAffinity;
+#elif __linux__
+	cpu_set_t dwProcessAffinity;
+#elif __OpenBSD__
+	int mib[2], core_count = 0;
+	int dwProcessAffinity = 0;
+	size_t len2;
+#else
+	int dwProcessAffinity = 0;
+	int32_t core_count = 0;
+#endif
+	size_t len = sizeof(dwProcessAffinity);
+	if (numthreads > 0)
+		return;
+#ifdef _WIN32
+	GetProcessAffinityMask(GetCurrentProcess(), &dwProcessAffinity, &dwSystemAffinity);
+#elif __linux__
+	sched_getaffinity(0, len, &dwProcessAffinity);
+#elif __OpenBSD__
+	len2 = sizeof(core_count);
+	mib[0] = CTL_HW;
+	mib[1] = HW_NCPU;
+	if (sysctl(mib, 2, &core_count, &len2, 0, 0) == 0)
+		numthreads = core_count;
+#else
+	if (sysctlbyname("hw.logicalcpu", &core_count, &len, 0, 0) == 0)
+		numthreads = core_count;
+	else if (sysctlbyname("hw.ncpu", &core_count, &len, 0, 0) == 0)
+		numthreads = core_count;
+#endif
+	for (unsigned int i = 0; i < len * 8; i++)
+#if defined(_WIN32)
+#if defined(_MSC_VER)
+		if (dwProcessAffinity & (1i64 << i))
+#else // CYGWIN/MINGW
+		if (dwProcessAffinity & (1ULL << i))
+#endif
+#elif defined __linux__
+		if (CPU_ISSET(i, &dwProcessAffinity))
+#else
+		if (dwProcessAffinity & (1 << i))
+#endif
+			numthreads++;
+	if (numthreads == 0) // something failed
+		numthreads = 1;
+	printf("Number of threads: %i\n", (int)numthreads);
+}
+
+extern "C" EXPORT unsigned long long BitmessagePOW(unsigned char * starthash, unsigned long long target)
+{
+	successval = 0;
+	max_val = target;
+	getnumthreads();
+	initialHash = (unsigned char *)starthash;
+#   ifdef _WIN32
+	HANDLE* threads = (HANDLE*)calloc(sizeof(HANDLE), numthreads);
+#   else
+	pthread_t* threads = (pthread_t*)calloc(sizeof(pthread_t), numthreads);
+	struct sched_param schparam;
+	schparam.sched_priority = 0;
+#   endif
+	unsigned int *threaddata = (unsigned int *)calloc(sizeof(unsigned int), numthreads);
+	for (unsigned int i = 0; i < numthreads; i++) {
+		threaddata[i] = i;
+#   ifdef _WIN32
+		threads[i] = CreateThread(NULL, 0, threadfunc, (LPVOID)&threaddata[i], 0, NULL);
+		SetThreadPriority(threads[i], THREAD_PRIORITY_IDLE);
+#   else
+		pthread_create(&threads[i], NULL, threadfunc, (void*)&threaddata[i]);
+#   ifdef __linux__
+		pthread_setschedparam(threads[i], SCHED_IDLE, &schparam);
+#   else
+		pthread_setschedparam(threads[i], SCHED_RR, &schparam);
+#   endif
+#   endif
+	}
+#   ifdef _WIN32
+	WaitForMultipleObjects(numthreads, threads, TRUE, INFINITE);
+#   else
+	for (unsigned int i = 0; i < numthreads; i++) {
+		pthread_join(threads[i], NULL);
+	}
+#   endif
+	free(threads);
+	free(threaddata);
+	return successval;
+}
+
+// python module definitions
+static PyObject *
+bitmessage_pow(PyObject *self, PyObject *args)
+{
+	const char *initial_hash;
+	unsigned long long target;
+	unsigned long long nonce;
+
+	if (!PyArg_ParseTuple(args, "yK", &initial_hash, target))
+		return NULL;
+	nonce = BitmessagePOW((unsigned char *)initial_hash, target);
+	return PyLong_FromUnsignedLongLong(nonce);
+};
+
+static PyMethodDef BitmsghashMethods[] = {
+	{"bitmessage_pow", bitmessage_pow, METH_VARARGS,
+	 "Do the Bitmessage PoW and return nonce."},
+	{NULL, NULL, 0, NULL}
+};
+
+static struct PyModuleDef bitmsghashmodule = {
+	PyModuleDef_HEAD_INIT,
+	"bitmsghash",
+	"A C extension for PoW borrowed from PyBitmessage",
+	-1,
+	BitmsghashMethods
+};
+
+PyMODINIT_FUNC
+PyInit_bitmsghash(void)
+{
+	return PyModule_Create(&bitmsghashmodule);
+}

minode/proofofwork.py

@@ -1,13 +1,17 @@
 """Doing proof of work"""
 import base64
+import ctypes
+import glob
 import hashlib
+import importlib
 import logging
 import multiprocessing
+import os
 import struct
 import threading
 import time
 
-from . import shared, structure
+from . import message, shared, structure
 
 
 def _pow_worker(target, initial_hash, q):
@@ -25,30 +29,103 @@ def _pow_worker(target, initial_hash, q):
     q.put(struct.pack('>Q', nonce))
 
 
-def _worker(obj):
+class Worker(threading.Thread):
-    q = multiprocessing.Queue()
+    """The thread doind PoW for objects and publishing them"""
-    p = multiprocessing.Process(
+    def __init__(self, obj=None):
-        target=_pow_worker, args=(obj.pow_target(), obj.pow_initial_hash(), q))
+        self.obj = obj
+        super().__init__(
+            name='Worker' if not self.obj
+            else 'Worker-%s' % (base64.b16encode(obj.vector).decode())
+        )
+        self.bmpow = bso = None
+        try:
+            bitmsglib = importlib.util.find_spec('minode.bitmsghash').origin
+            if not bitmsglib:
+                bitmsglib = glob.glob(os.path.join(
+                    os.path.dirname(os.path.abspath(__file__)),
+                    'bitmsghash*.so'
+                ))[0]
+        except (AttributeError, IndexError):
+            return
 
-    logging.debug('Starting POW process')
+        try:
-    t = time.time()
+            bso = ctypes.CDLL(bitmsglib)
-    p.start()
+        except Exception:  # pylint: disable=broad-exception-caught
-    nonce = q.get()
+            logging.warning('Error loading bitmsghash', exc_info=True)
-    p.join()
+            try:
+                # MSVS on Windows
+                bso = ctypes.WinDLL(bitmsglib)
+            except (AttributeError, TypeError, ValueError):
+                return
 
-    logging.debug(
+        self.bmpow = bso.BitmessagePOW
-        'Finished doing POW, nonce: %s, time: %ss', nonce, time.time() - t)
+        self.bmpow.restype = ctypes.c_ulonglong
-    obj = structure.Object(
-        nonce, obj.expires_time, obj.object_type, obj.version,
-        obj.stream_number, obj.object_payload)
-    logging.debug(
-        'Object vector is %s', base64.b16encode(obj.vector).decode())
 
-    with shared.objects_lock:
+    @staticmethod
-        shared.objects[obj.vector] = obj
+    def _worker_result(obj):
-        shared.vector_advertise_queue.put(obj.vector)
+        q = multiprocessing.Queue()
+        p = multiprocessing.Process(target=_pow_worker, args=(
+            obj.pow_target(), obj.pow_initial_hash(), q))
+
+        p.start()
+        nonce = q.get()
+        p.join()
+
+        return nonce
+
+    def add_pow(self, obj):
+        """
+        Do PoW for the given object and return a new `.structure.Object`
+        - with the proper nonce and vector.
+        """
+        t = time.time()
+        if not self.bmpow:
+            logging.debug('Starting PoW process')
+            nonce = self._worker_result(obj)
+        else:
+            logging.debug('Calling PoW extension')
+            nonce = struct.pack('>Q', self.bmpow(
+                ctypes.pointer(
+                    ctypes.create_string_buffer(obj.pow_initial_hash(), 64)),
+                ctypes.c_ulonglong(obj.pow_target())
+            ))
+
+        logging.debug(
+            'Finished doing PoW, nonce: %s, time: %ss', nonce, time.time() - t)
+        obj = structure.Object(
+            nonce, obj.expires_time, obj.object_type, obj.version,
+            obj.stream_number, obj.object_payload)
+        logging.debug(
+            'Object vector is %s', base64.b16encode(obj.vector).decode())
+
+        return obj
+
+    @staticmethod
+    def _publish(obj):
+        with shared.objects_lock:
+            shared.objects[obj.vector] = obj
+            shared.vector_advertise_queue.put(obj.vector)
+
+    def run(self):
+        if self.obj:
+            self._publish(self.add_pow(self.obj))
+            return
+
+        while not shared.shutting_down:
+            data = shared.objects_queue.get()
+
+            obj = structure.Object.from_message(
+                message.Message.from_bytes(data))
+            if int.from_bytes(obj.nonce, 'big') == 0:
+                obj = self.add_pow(obj)
+
+            self._publish(obj)
 
 
 def do_pow_and_publish(obj):
-    t = threading.Thread(target=_worker, args=(obj, ))
+    """
-    t.start()
+    Start a worker thread doing PoW for the given object
+    and putting a new object and its vector into appropriate places in `shared`
+    to advertize to the network.
+    """
+    Worker(obj).start()

minode/shared.py

@@ -64,3 +64,5 @@ connection_limit = 250
 
 objects = {}
 objects_lock = threading.Lock()
+
+objects_queue = queue.Queue()

minode/tests/test_structure.py

@@ -122,8 +122,7 @@ class TestStructure(unittest.TestCase):
         obj = structure.Object(
             b'\x00' * 8, int(time.time() + 300), 42, 1, 2, b'HELLO')
         vector = obj.vector
-        proofofwork._worker(obj)  # pylint: disable=protected-access
+        obj = proofofwork.Worker().add_pow(obj)
-        obj = shared.objects.popitem()[1]
         self.assertNotEqual(obj.vector, vector)
         self.assertFalse(obj.is_expired())
         self.assertFalse(obj.is_valid())

setup.py

@@ -1,8 +1,9 @@
 #!/usr/bin/env python
 
 import os
+import sys
 
-from setuptools import setup, find_packages
+from setuptools import Extension, find_packages, setup
 
 from minode import shared
 
@@ -12,6 +13,19 @@ README = open(os.path.join(
 
 name, version = shared.user_agent.strip(b'/').split(b':')
 
+bitmsghash = Extension(
+    'minode.bitmsghash',
+    sources=['minode/bitmsghash/bitmsghash.cpp'],
+    libraries=['pthread', 'crypto'],
+)
+
+if sys.platform[:3] == 'win':
+    bitmsghash.libraries = ['libeay32', 'ws2_32']
+    openssl_dir = os.getenv('OPENSSL_DIR')
+    if openssl_dir:
+        bitmsghash.library_dirs = [os.path.join(openssl_dir, 'lib')]
+        bitmsghash.include_dirs = [os.path.join(openssl_dir, 'include')]
+
 setup(
     name=name.decode('utf-8'),
     version=version.decode('utf-8'),
@@ -23,6 +37,7 @@ setup(
     url='https://git.bitmessage.org/lee.miller/MiNode',
     packages=find_packages(exclude=('*tests',)),
     package_data={'': ['*.csv', 'tls/*.pem']},
+    ext_modules=[bitmsghash],
     entry_points={'console_scripts': ['minode = minode.main:main']},
     classifiers=[
         "License :: OSI Approved :: MIT License"

tox.ini

@@ -12,6 +12,9 @@ deps = flake8
 commands =
     flake8 minode --count --select=E9,F63,F7,F82 --show-source --statistics
 
+[testenv:py39]
+commands_pre = pip install -e .
+
 [testenv:reset]
 deps =
     -rrequirements.txt