Backups/multi-user
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(rcf): apprend basic multithreaded client api

Browse Source
This commit is contained in:
Swann
2019-04-08 15:54:21 +02:00
parent fecc429ef1
commit c137971606
6 changed files with 163 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

272
net_components.py
View File

@ -1,11 +1,17 @@
import collections import collections
import logging import logging
import threading
from uuid import uuid4 from uuid import uuid4
import binascii
import os
from random import randint
import time import time
from enum import Enum from enum import Enum
from .libs import umsgpack, zmq try:
from .libs import umsgpack, zmq
except:
from libs import umsgpack, zmq
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
@ -13,6 +19,7 @@ CONNECT_TIMEOUT = 2
WAITING_TIME = 0.001 WAITING_TIME = 0.001
SERVER_MAX = 1 SERVER_MAX = 1
def zpipe(ctx): def zpipe(ctx):
"""build inproc pipe for talking to threads """build inproc pipe for talking to threads
@ -27,7 +34,7 @@ def zpipe(ctx):
iface = "inproc://%s" % binascii.hexlify(os.urandom(8)) iface = "inproc://%s" % binascii.hexlify(os.urandom(8))
a.bind(iface) a.bind(iface)
b.connect(iface) b.connect(iface)
return a,b return a, b
class State(Enum): class State(Enum):
@ -36,52 +43,10 @@ class State(Enum):
ACTIVE = 3 ACTIVE = 3
class RCFFactory(object):
"""
Abstract layer used to bridge external and inter
"""
def init(self, data):
"""
set the RCFMessage pointer to local data
"""
print("Default setter")
# Setup data accessor
data.get = self.load_getter(data)
data.set = self.load_setter(data)
# TODO: Setup local pointer
def load_getter(self, data):
"""
local program > rcf
"""
print("Default getter")
return None
def load_setter(self, data):
"""
rcf > local program
"""
print("Default setter")
return None
def apply(self, data):
pass
def diff(self, data):
"""
Verify data integrity
"""
pass
class RCFStore(collections.MutableMapping, dict): class RCFStore(collections.MutableMapping, dict):
def __init__(self, custom_factory=RCFFactory()): def __init__(self):
super().__init__() super().__init__()
self.factory = custom_factory
def __getitem__(self, key): def __getitem__(self, key):
return dict.__getitem__(self, key) return dict.__getitem__(self, key)
@ -117,22 +82,20 @@ class RCFMessage(object):
body = None # data blob body = None # data blob
uuid = None uuid = None
def __init__(self, key=None,uuid= None, id=None, mtype=None, body=None): def __init__(self, key=None, uuid=None, id=None, mtype=None, body=None):
if uuid is None: if uuid is None:
uuid = uuid4() uuid = uuid4().bytes
self.key = key self.key = key
self.uuid = uuid self.uuid = uuid
self.mtype = mtype self.mtype = mtype
self.body = body self.body = body
self.id = id self.id = id
def store(self, dikt): def store(self, dikt):
"""Store me in a dict if I have anything to store""" """Store me in a dict if I have anything to store"""
# this currently erasing old value # this currently erasing old value
if self.key is not None : if self.key is not None:
dikt[self.key] = self dikt[self.key] = self
# elif self.key in dikt: # elif self.key in dikt:
# del dikt[self.key] # del dikt[self.key]
@ -140,26 +103,25 @@ class RCFMessage(object):
def send(self, socket): def send(self, socket):
"""Send key-value message to socket; any empty frames are sent as such.""" """Send key-value message to socket; any empty frames are sent as such."""
key = ''.encode() if self.key is None else self.key.encode() key = ''.encode() if self.key is None else self.key.encode()
print(self.mtype)
mtype = ''.encode() if self.mtype is None else self.mtype.encode() mtype = ''.encode() if self.mtype is None else self.mtype.encode()
body = ''.encode() if self.body is None else umsgpack.packb(self.body) body = ''.encode() if self.body is None else umsgpack.packb(self.body)
id = ''.encode() if self.id is None else self.id id = ''.encode() if self.id is None else self.id
try: try:
socket.send_multipart([key,self.uuid, id, mtype, body]) socket.send_multipart([key, id, mtype, body])
except: except:
logger.info("Fail to send {}".format(key)) logger.info("Fail to send {} {}".format(key,id))
@classmethod @classmethod
def recv(cls, socket): def recv(cls, socket):
"""Reads key-value message from socket, returns new kvmsg instance.""" """Reads key-value message from socket, returns new kvmsg instance."""
key,uuid, id, mtype, body = socket.recv_multipart(zmq.DONTWAIT) key, id, mtype, body = socket.recv_multipart(zmq.DONTWAIT)
key = key.decode() if key else None key = key.decode() if key else None
id = id if id else None id = id if id else None
mtype = mtype.decode() if body else None mtype = mtype.decode() if body else None
body = umsgpack.unpackb(body) if body else None body = umsgpack.unpackb(body) if body else None
return cls(key=key,uuid=uuid, id=id, mtype=mtype, body=body) return cls(key=key, id=id, mtype=mtype, body=body)
def dump(self): def dump(self):
if self.body is None: if self.body is None:
@ -176,84 +138,29 @@ class RCFMessage(object):
)) ))
class RCFClient(object):
ctx = None
pipe = None
agent = None
class RCFClient(): def __init__(self):
def __init__( self.ctx = zmq.Context()
self, self.pipe, peer = zpipe(self.ctx)
context=zmq.Context(), self.agent = threading.Thread(
id="default", target=rcf_client_agent, args=(self.ctx, peer))
on_recv=None, self.agent.daemon = True
on_post_init=None, self.agent.start()
is_admin=False,
factory=None,
address="localhost"):
# 0MQ vars def connect(self, address, port):
self.context = context self.pipe.send_multipart([b"CONNECT", (address.encode() if isinstance(
self.pull_sock = None address, str) else address), b'%d' % port])
self.req_sock = None
self.poller = None
# Client configuration def set(self, key, value):
self.id = id.encode() """Set new value in distributed hash table
self.on_recv = on_recv Sends [SET][key][value][ttl] to the agent
self.on_post_init = on_post_init """
self.status = RCFStatus.IDLE self.pipe.send_multipart([b"SET", umsgpack.packb(key), umsgpack.packb(value)])
self.is_admin = is_admin
self.address = address
self.bind_ports()
# client routine registration
self.load_task = asyncio.ensure_future(self.load())
self.tick_task = None
logger.info("{} client initialized".format(id))
def bind_ports(self):
# pull socket: get update FROM server
self.pull_sock = self.context.socket(zmq.SUB)
self.pull_sock.linger = 0
self.pull_sock.connect("tcp://{}:5555".format(self.address))
self.pull_sock.setsockopt_string(zmq.SUBSCRIBE, '')
# request socket: send request/message over all peers throught the server
self.req_sock = self.context.socket(zmq.DEALER)
self.req_sock.setsockopt(zmq.IDENTITY, self.id)
# self.req_sock.setsockopt(zmq.SNDHWM, 60)
self.req_sock.linger = 0
self.req_sock.connect("tcp://{}:5556".format(self.address))
# push update socket
self.push_sock = self.context.socket(zmq.PUSH)
self.push_sock.setsockopt(zmq.IDENTITY, self.id)
self.push_sock.linger = 0
self.push_sock.connect("tcp://{}:5557".format(self.address))
self.push_sock.setsockopt(zmq.SNDHWM, 60)
# Sockets aggregator, not really used for now
self.poller = zmq.Poller()
self.poller.register(self.pull_sock, zmq.POLLIN)
time.sleep(0.1)
def push_update(self, key, mtype, body):
rcfmsg = RCFMessage(key=key, id=self.id,mtype=mtype, body=body)
rcfmsg.send(self.push_sock)
def stop(self):
logger.debug("Stopping client")
self.poller.unregister(self.pull_sock)
self.req_sock.close()
self.push_sock.close()
self.pull_sock.close()
self.load_task.cancel()
if self.tick_task:
self.tick_task.cancel()
class RCFServer(object): class RCFServer(object):
address = None # Server address address = None # Server address
@ -261,19 +168,22 @@ class RCFServer(object):
snapshot = None # Snapshot socket snapshot = None # Snapshot socket
subscriber = None # Incoming updates subscriber = None # Incoming updates
def __init__(self, ctx, address, port): def __init__(self, ctx, address, port,id):
self.address = address self.address = address
self.port = port self.port = port
self.snapshot = ctx.socket(zmq.DEALER) self.snapshot = ctx.socket(zmq.DEALER)
self.snapshot.linger = 0 self.snapshot.linger = 0
self.snapshot.connect("%s:%i".format(address.decode(),port)) self.snapshot.connect("tcp://{}:{}".format(address.decode(), port))
self.snapshot.setsockopt(zmq.IDENTITY, id)
self.subscriber = ctx.socket(zmq.SUB) self.subscriber = ctx.socket(zmq.SUB)
self.subscriber.setsockopt_string(zmq.SUBSCRIBE, '') self.subscriber.setsockopt_string(zmq.SUBSCRIBE, '')
self.subscriber.connect("%s:%i".format(address.decode(),port+1)) self.subscriber.connect("tcp://{}:{}".format(address.decode(), port+1))
self.subscriber.linger = 0 self.subscriber.linger = 0
print("connected on tcp://{}:{}".format(address.decode(), port))
class RCFClientAgent(object): class RCFClientAgent(object):
ctx = None ctx = None
pipe = None pipe = None
property_map = None property_map = None
publisher = None publisher = None
@ -281,21 +191,19 @@ class RCFClientAgent(object):
state = State.INITIAL state = State.INITIAL
server = None server = None
def __init__(self, ctx, pipe, id): def __init__(self, ctx, pipe):
self.ctx = None self.ctx = ctx
self.pipe = None self.pipe = pipe
self.property_map = None self.property_map = RCFStore()
self.publisher = None self.id = b"test"
self.id = None
self.state = State.INITIAL self.state = State.INITIAL
self.server = None self.server = None
self.publisher = self.context.socket(zmq.PUSH) # push update socket self.publisher = self.ctx.socket(zmq.PUSH) # push update socket
self.publisher.setsockopt(zmq.IDENTITY, self.id) self.publisher.setsockopt(zmq.IDENTITY, self.id)
self.publisher.setsockopt(zmq.SNDHWM, 60) self.publisher.setsockopt(zmq.SNDHWM, 60)
self.publisher.linger = 0 self.publisher.linger = 0
def control_message (self): def control_message(self):
msg = self.pipe.recv_multipart() msg = self.pipe.recv_multipart()
command = msg.pop(0) command = msg.pop(0)
@ -303,32 +211,43 @@ class RCFClientAgent(object):
address = msg.pop(0) address = msg.pop(0)
port = int(msg.pop(0)) port = int(msg.pop(0))
if len(self.servers) < SERVER_MAX: if self.server is None:
self.server = RCFServer(self.ctx, address, port) self.server = RCFServer(self.ctx, address, port, self.id)
self.publisher.connect("tcp://{}:5557".format(address.decode())) self.publisher.connect("tcp://{}:{}".format(address.decode(), port+2))
else: else:
logger.error("E: too many servers (max. %i)", SERVER_MAX) logger.error("E: too many servers (max. %i)", SERVER_MAX)
elif command == b"SET":
key,value = msg
# Send key-value pair on to server
rcfmsg = RCFMessage(key=umsgpack.unpackb(key),id=self.id ,mtype="",body=umsgpack.unpackb(value))
rcfmsg.store(self.property_map)
rcfmsg.send(self.publisher)
def rcf_client_agent(ctx,pipe,id): def rcf_client_agent(ctx, pipe):
agent = RCFClientAgent(ctx,pipe,id) agent = RCFClientAgent(ctx, pipe)
server = None server = None
while True: while True:
# logger.info("asdasd")
poller = zmq.Poller() poller = zmq.Poller()
poller.register(agent.pipe, zmq.POLLIN) poller.register(agent.pipe, zmq.POLLIN)
server_socket = None server_socket = None
if agent.state == State.INITIAL: if agent.state == State.INITIAL:
server = agent.server server = agent.server
if agent.servers: if agent.server:
logger.info ("I: waiting for server at %s:%d...", logger.info("I: waiting for server at %s:%d...",
server.address, server.port) server.address, server.port)
server.snapshot.send(b"SNAPSHOT_REQUEST") server.snapshot.send(b"SNAPSHOT_REQUEST")
agent.state = State.SYNCING agent.state = State.SYNCING
server_socket = server.snapshot
elif agent.state == State.SYNCING: elif agent.state == State.SYNCING:
sever_socket = server.snapshot server_socket = server.snapshot
elif agent.state == State.ACTIVE: elif agent.state == State.ACTIVE:
server_socket = server.subscriber server_socket = server.subscriber
@ -338,14 +257,12 @@ def rcf_client_agent(ctx,pipe,id):
try: try:
items = dict(poller.poll()) items = dict(poller.poll())
except: except:
raise pass
break
if agent.pipe in items: if agent.pipe in items:
agent.control_message() agent.control_message()
elif server_socket in items: elif server_socket in items:
rcfmsg = RCFMessage.recv(server_socket) rcfmsg = RCFMessage.recv(server_socket)
if agent.state == State.SYNCING: if agent.state == State.SYNCING:
# Store snapshot # Store snapshot
if rcfmsg.key == "SNAPSHOT_END": if rcfmsg.key == "SNAPSHOT_END":
@ -356,16 +273,15 @@ def rcf_client_agent(ctx,pipe,id):
elif agent.state == State.ACTIVE: elif agent.state == State.ACTIVE:
if rcfmsg.id != agent.id: if rcfmsg.id != agent.id:
rcfmsg.store(agent.property_map) rcfmsg.store(agent.property_map)
action = "update" if kvmsg.body else "delete" action = "update" if rcfmsg.body else "delete"
logging.info ("I: received from %s:%d %s", logging.info("I: received from {}:{},{} {}".format(server.address,rcfmsg.body.id, server.port, action))
server.address, server.port, action) else:
else: logger.info("IDLE")
agent.state = State.INITIAL # else: else
# agent.state = State.INITIAL
class RCFServerAgent(): class RCFServerAgent():
def __init__(self, context=zmq.Context(), id="admin"): def __init__(self, context=zmq.Context.instance(), id="admin"):
self.context = context self.context = context
self.pub_sock = None self.pub_sock = None
@ -373,11 +289,11 @@ class RCFServerAgent():
self.collector_sock = None self.collector_sock = None
self.poller = None self.poller = None
self.property_map = RCFStore() self.property_map = {}
self.id = id self.id = id
self.bind_ports() self.bind_ports()
# Main client loop registration # Main client loop registration
tick() self.tick()
logger.info("{} client initialized".format(id)) logger.info("{} client initialized".format(id))
@ -385,14 +301,14 @@ class RCFServerAgent():
# Update all clients # Update all clients
self.pub_sock = self.context.socket(zmq.PUB) self.pub_sock = self.context.socket(zmq.PUB)
self.pub_sock.setsockopt(zmq.SNDHWM, 60) self.pub_sock.setsockopt(zmq.SNDHWM, 60)
self.pub_sock.bind("tcp://*:5555") self.pub_sock.bind("tcp://*:5556")
time.sleep(0.2) time.sleep(0.2)
# Update request # Update request
self.request_sock = self.context.socket(zmq.ROUTER) self.request_sock = self.context.socket(zmq.ROUTER)
self.request_sock.setsockopt(zmq.IDENTITY, b'SERVER') self.request_sock.setsockopt(zmq.IDENTITY, b'SERVER')
self.request_sock.setsockopt(zmq.RCVHWM, 60) self.request_sock.setsockopt(zmq.RCVHWM, 60)
self.request_sock.bind("tcp://*:5556") self.request_sock.bind("tcp://*:5555")
# Update collector # Update collector
self.collector_sock = self.context.socket(zmq.PULL) self.collector_sock = self.context.socket(zmq.PULL)
@ -409,7 +325,7 @@ class RCFServerAgent():
while True: while True:
# Non blocking poller # Non blocking poller
socks = dict(self.poller.poll()) socks = dict(self.poller.poll(1000))
# Snapshot system for late join (Server - Client) # Snapshot system for late join (Server - Client)
if self.request_sock in socks: if self.request_sock in socks:
@ -417,7 +333,7 @@ class RCFServerAgent():
identity = msg[0] identity = msg[0]
request = msg[1] request = msg[1]
print("asdasd")
if request == b"SNAPSHOT_REQUEST": if request == b"SNAPSHOT_REQUEST":
pass pass
else: else:
@ -437,16 +353,8 @@ class RCFServerAgent():
# Regular update routing (Clients / Client) # Regular update routing (Clients / Client)
elif self.collector_sock in socks: elif self.collector_sock in socks:
msg = RCFMessage.recv(self.collector_sock) msg = RCFMessage.recv(self.collector_sock)
# Update all clients # Update all clients
msg.store(self.property_map) msg.store(self.property_map)
msg.send(self.pub_sock) msg.send(self.pub_sock)
def stop(self):
logger.debug("Stopping server")
self.poller.unregister(self.request_sock)
self.pub_sock.close()
self.request_sock.close()
self.collector_sock.close()
self.status = RCFStatus.IDLE

49
net_operators.py
View File

@ -672,8 +672,8 @@ class session_join(bpy.types.Operator):
net_settings = context.scene.session_settings net_settings = context.scene.session_settings
# Scene setup # Scene setup
if net_settings.session_mode == "CONNECT" and net_settings.clear_scene: # if net_settings.session_mode == "CONNECT" and net_settings.clear_scene:
clean_scene() # clean_scene()
# Session setup # Session setup
if net_settings.username == "DefaultUser": if net_settings.username == "DefaultUser":
@ -682,23 +682,21 @@ class session_join(bpy.types.Operator):
username = str(context.scene.session_settings.username) username = str(context.scene.session_settings.username)
client = net_components.RCFClient(
id=username,
on_recv=recv_callbacks,
on_post_init=post_init_callbacks,
address=net_settings.ip,
is_admin=net_settings.session_mode == "HOST")
bpy.ops.asyncio.loop() client = net_components.RCFClient()
client.connect("127.0.0.1",5555)
client.set('key', 1)
net_settings.is_running = True
drawer = net_draw.HUD(client_instance=client) # net_settings.is_running = True
register_ticks() # drawer = net_draw.HUD(client_instance=client)
# register_ticks()
return {"FINISHED"} return {"FINISHED"}
class session_add_property(bpy.types.Operator): class session_add_property(bpy.types.Operator):
bl_idname = "session.add_prop" bl_idname = "session.add_prop"
bl_label = "add" bl_label = "add"
@ -715,21 +713,22 @@ class session_add_property(bpy.types.Operator):
def execute(self, context): def execute(self, context):
global client global client
item = resolve_bpy_path(self.property_path) client.set('key', 1)
# item = resolve_bpy_path(self.property_path)
print(item) # print(item)
if item: # if item:
key = self.property_path # key = self.property_path
dumper = dump_anything.Dumper() # dumper = dump_anything.Dumper()
dumper.type_subset = dumper.match_subset_all # dumper.type_subset = dumper.match_subset_all
dumper.depth = self.depth # dumper.depth = self.depth
data = dumper.dump(item) # data = dumper.dump(item)
data_type = item.__class__.__name__ # data_type = item.__class__.__name__
client.push_update(key, data_type, data) # client.push_update(key, data_type, data)
return {"FINISHED"} return {"FINISHED"}
@ -771,7 +770,7 @@ class session_create(bpy.types.Operator):
global server global server
global client global client
server = net_components.RCFServer() server = net_components.RCFServerAgent()
time.sleep(0.1) time.sleep(0.1)
bpy.ops.session.join() bpy.ops.session.join()
@ -983,11 +982,11 @@ def unregister():
pass pass
if server: if server:
server.stop() # server.stop()
del server del server
server = None server = None
if client: if client:
client.stop() # client.stop()
del client del client
client = None client = None

26
net_ui.py
View File

@ -55,17 +55,17 @@ class SessionSettingsPanel(bpy.types.Panel):
row = layout.row() row = layout.row()
row.operator("session.join", text="CONNECT") row.operator("session.join", text="CONNECT")
else: # else:
if net_operators.client.status is net_components.RCFStatus.CONNECTED: # if net_operators.client.status is net_components.RCFStatus.CONNECTED:
row.label(text="Net frequency:") # row.label(text="Net frequency:")
row.prop(net_settings, "update_frequency", text="") # row.prop(net_settings, "update_frequency", text="")
row = layout.row() # row = layout.row()
row.operator("session.stop", icon='QUIT', text="Exit") # row.operator("session.stop", icon='QUIT', text="Exit")
elif net_operators.client.status is net_components.RCFStatus.CONNECTING: # elif net_operators.client.status is net_components.RCFStatus.CONNECTING:
row.label(text="connecting...") # row.label(text="connecting...")
row = layout.row() # row = layout.row()
row.operator("session.stop", icon='QUIT', text="CANCEL") # row.operator("session.stop", icon='QUIT', text="CANCEL")
row = layout.row() row = layout.row()
@ -198,9 +198,9 @@ class SessionTaskPanel(bpy.types.Panel):
classes = ( classes = (
SessionSettingsPanel, SessionSettingsPanel,
SessionUsersPanel, # SessionUsersPanel,
SessionPropertiesPanel, # SessionPropertiesPanel,
SessionTaskPanel, # SessionTaskPanel,
) )

20
rcf_server.py Normal file
View File

@ -0,0 +1,20 @@
import collections
import logging
import threading
from uuid import uuid4
import binascii
import os
from random import randint
import time
from enum import Enum
from libs import umsgpack, zmq
from net_components import RCFMessage
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG)
CONNECT_TIMEOUT = 2
WAITING_TIME = 0.001
SERVER_MAX = 1

14
test_client.py Normal file
View File

@ -0,0 +1,14 @@
from net_components import RCFClient
import time
client = RCFClient()
client.connect("127.0.0.1",5555)
try:
while True:
client.set('key', 1)
# Distribute as key-value message
time.sleep(1)
except KeyboardInterrupt:
pass

3
test_server.py
View File

@ -1,3 +1,4 @@
from net_components import RCFServerAgent from net_components import RCFServerAgent
server = RCFServerAgent() server = RCFServerAgent()