Work on new gossip system

This commit is contained in:
Kevin F 2022-02-11 00:56:19 -06:00
parent ebb75b136d
commit 026f39b680
6 changed files with 65 additions and 30 deletions

View File

@ -1,9 +1,14 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Set
from gossip.peer import Peer
if TYPE_CHECKING: if TYPE_CHECKING:
import queue import queue
from onionrblocks import Block from onionrblocks import Block
from onionrthreads import add_onionr_thread from onionrthreads import add_onionr_thread
import onionrplugins
from .client import gossip_client
from .server import gossip_server
""" """
Onionr uses a flavor of Dandelion++ epidemic routing Onionr uses a flavor of Dandelion++ epidemic routing
@ -20,9 +25,12 @@ When a new block is generated, it is added to a queue in raw form passed to the
""" """
def start_gossip_threads(block_queue: queue.Queue[Block]): def start_gossip_threads(peer_set: Set[Peer], block_queue: queue.Queue[Block]):
# Peer set is largely handled by the transport plugins # Peer set is largely handled by the transport plugins
# There is a unified set so gossip logic is not repeated # There is a unified set so gossip logic is not repeated
peer_set = set()
add_onionr_thread(gossip_server, 1, peer_set, block_queue, initial_sleep=0.2)
add_onionr_thread(gossip_client, 1, peer_set, block_queue, initial_sleep=0)
onionrplugins.events.event('gossip_start', data=peer_set, threaded=True)

View File

@ -0,0 +1,12 @@
from typing import TYPE_CHECKING
from typing import Set
from queue import Queue
if TYPE_CHECKING:
from onionrblocks import Block
from peer import Peer
def gossip_client(peer_set: Set[Peer], block_queue: Queue[Block]):
return

View File

@ -0,0 +1,12 @@
from typing import TYPE_CHECKING
from typing import Set
from queue import Queue
if TYPE_CHECKING:
from onionrblocks import Block
from peer import Peer
def gossip_server(peer_set: Set[Peer], block_queue: Queue[Block]):
return

View File

@ -33,6 +33,7 @@ from .. import version
from utils.bettersleep import better_sleep from utils.bettersleep import better_sleep
from .killdaemon import kill_daemon # noqa from .killdaemon import kill_daemon # noqa
from .showlogo import show_logo from .showlogo import show_logo
import gossip
from setupkvvars import setup_kv from setupkvvars import setup_kv
""" """
@ -117,6 +118,8 @@ def daemon():
events.event('init', threaded=False) events.event('init', threaded=False)
events.event('daemon_start') events.event('daemon_start')
gossip.start_gossip_threads(shared_state.get(DeadSimpleKV)['peers'], shared_state.get(DeadSimpleKV)['block_queue'])
try: try:
shared_state.get(apiservers.ClientAPI).start() shared_state.get(apiservers.ClientAPI).start()
except KeyboardInterrupt: except KeyboardInterrupt:

View File

@ -1,9 +1,14 @@
''' """
Onionr - Private P2P Communication Onionr - Private P2P Communication
This file deals with configuration management. Deals with configuration management.
''' """
''' from threading import Thread
import config, logger
import onionrplugins as plugins
from . import onionrpluginapi as pluginapi
"""
This program is free software: you can redistribute it and/or modify This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or the Free Software Foundation, either version 3 of the License, or
@ -16,22 +21,18 @@
You should have received a copy of the GNU General Public License You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>. along with this program. If not, see <https://www.gnu.org/licenses/>.
''' """
from threading import Thread
import config, logger
import onionrplugins as plugins
from . import onionrpluginapi as pluginapi
def get_pluginapi(data): def get_pluginapi(data):
return pluginapi.SharedAPI(data) return pluginapi.SharedAPI(data)
def __event_caller(event_name, data = {}): def __event_caller(event_name, data = {}):
''' """DO NOT call this function, this is for threading code only.
DO NOT call this function, this is for threading code only.
Instead, call onionrevents.event Instead, call onionrevents.event
''' """
disabled = config.get('plugins.disabled') disabled = config.get('plugins.disabled')
for plugin in plugins.get_enabled_plugins(): for plugin in plugins.get_enabled_plugins():
if plugin in disabled: continue if plugin in disabled: continue
@ -45,9 +46,7 @@ def __event_caller(event_name, data = {}):
logger.debug((event_name + ' - ' + plugin + ' - ' + str(e)), terminal=True) logger.debug((event_name + ' - ' + plugin + ' - ' + str(e)), terminal=True)
def event(event_name, data = {}, threaded = True): def event(event_name, data = {}, threaded = True):
''' """Call an event on all plugins (if defined)"""
Calls an event on all plugins (if defined)
'''
if threaded: if threaded:
thread = Thread(target = __event_caller, args = (event_name, data)) thread = Thread(target = __event_caller, args = (event_name, data))
@ -57,9 +56,7 @@ def event(event_name, data = {}, threaded = True):
__event_caller(event_name, data) __event_caller(event_name, data)
def call(plugin, event_name, data = None, pluginapi = None): def call(plugin, event_name, data = None, pluginapi = None):
''' """Call an event on a plugin if one is defined"""
Calls an event on a plugin if one is defined
'''
if not plugin is None: if not plugin is None:
try: try:

View File

@ -2,6 +2,7 @@
Initialize singleton deadsimplekv pseudo globals Initialize singleton deadsimplekv pseudo globals
""" """
import queue
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@ -27,6 +28,8 @@ along with this program. If not, see <https://www.gnu.org/licenses/>.
def setup_kv(shared_vars: 'DeadSimpleKV'): def setup_kv(shared_vars: 'DeadSimpleKV'):
"""Init initial pseudo-globals.""" """Init initial pseudo-globals."""
shared_vars.put("peers", set())
shared_vars.put("block_queue", queue.Queue())
shared_vars.put('shutdown', False) shared_vars.put('shutdown', False)
shared_vars.put('generating_blocks', []) shared_vars.put('generating_blocks', [])
shared_vars.put('startTime', epoch.get_epoch()) shared_vars.put('startTime', epoch.get_epoch())