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

feat: apply logic

Browse Source 
feat: state draft
This commit is contained in:
Swann Martinez
2019-07-18 16:38:13 +02:00
parent aa73064eed
commit 51093b1307
4 changed files with 208 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
client.py
View File

@ -52,7 +52,6 @@ def zpipe(ctx):
class Client(object): class Client(object):
ctx = None ctx = None
pipe = None pipe = None
net_agent = None net_agent = None
@ -205,7 +204,7 @@ class Client(object):
# SAVING FUNCTIONS # SAVING FUNCTIONS
def dump(self, filepath): def dump(self, filepath):
with open('dump.json',"w") as fp: with open('dump.json', "w") as fp:
for key, value in self.store.items(): for key, value in self.store.items():
line = json.dumps(value.body) line = json.dumps(value.body)
fp.write(line) fp.write(line)

106
replication.py
View File

@ -8,23 +8,36 @@ except:
from libs import umsgpack from libs import umsgpack
import zmq import zmq
import pickle
from enum import Enum
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
class RepState(Enum):
ADDED = 0
COMMITED = 1
STAGED = 2
class ReplicatedDataFactory(object): class ReplicatedDataFactory(object):
""" """
Manage the data types implamentations Manage the data types implementations
""" """
def __init__(self): def __init__(self):
self.supported_types = [] self.supported_types = []
def register_type(self,dtype, implementation): # Default registered types
self.register_type(str,RepCommand)
def register_type(self, dtype, implementation):
""" """
Register a new replicated datatype implementation Register a new replicated datatype implementation
""" """
self.supported_types.append((dtype, implementation)) self.supported_types.append((dtype, implementation))
def match_type_by_instance(self,data): def match_type_by_instance(self, data):
""" """
Find corresponding type to the given datablock Find corresponding type to the given datablock
""" """
@ -35,14 +48,14 @@ class ReplicatedDataFactory(object):
print("type not supported for replication") print("type not supported for replication")
raise NotImplementedError raise NotImplementedError
def match_type_by_name(self,type_name): def match_type_by_name(self, type_name):
for stypes, implementation in self.supported_types: for stypes, implementation in self.supported_types:
if type_name == stypes.__name__: if type_name == implementation.__name__:
return implementation return implementation
print("type not supported for replication")
raise NotImplementedError
return None def construct_from_dcc(self, data):
def construct_from_dcc(self,data):
implementation = self.match_type_by_instance(data) implementation = self.match_type_by_instance(data)
return implementation return implementation
@ -52,24 +65,26 @@ class ReplicatedDataFactory(object):
""" """
return self.match_type_by_name(type_name) return self.match_type_by_name(type_name)
class ReplicatedDatablock(object): class ReplicatedDatablock(object):
""" """
Datablock used for replication Datablock used for replication
""" """
uuid = None # key (string) uuid = None # uuid used as key (string)
pointer = None # dcc data reference pointer = None # dcc data ref (DCC type)
data = None # data blob (json) buffer = None # data blob (json)
str_type = None # data type name (for deserialization) str_type = None # data type name (string)
deps = None # dependencies references deps = [None] # dependencies array (string)
owner = None owner = None # Data owner (string)
state = None # Data state (RepState)
def __init__(self, owner=None, data=None, uuid=None): def __init__(self, owner=None, data=None, uuid=None, buffer=None):
self.uuid = uuid if uuid else str(uuid4()) self.uuid = uuid if uuid else str(uuid4())
assert(owner) assert(owner)
self.owner = owner self.owner = owner
self.pointer = data self.pointer = data
self.str_type = type(data).__name__ self.buffer = buffer if buffer else None
self.str_type = type(self).__name__
def push(self, socket): def push(self, socket):
""" """
@ -83,7 +98,7 @@ class ReplicatedDatablock(object):
key = self.uuid.encode() key = self.uuid.encode()
type = self.str_type.encode() type = self.str_type.encode()
socket.send_multipart([key,owner,type,data]) socket.send_multipart([key, owner, type, data])
@classmethod @classmethod
def pull(cls, socket, factory): def pull(cls, socket, factory):
@ -92,53 +107,80 @@ class ReplicatedDatablock(object):
- read data from the socket - read data from the socket
- reconstruct an instance - reconstruct an instance
""" """
uuid, owner,str_type, data = socket.recv_multipart(zmq.NOBLOCK) uuid, owner, str_type, data = socket.recv_multipart(zmq.NOBLOCK)
str_type = str_type.decode() str_type = str_type.decode()
owner=owner.decode() owner = owner.decode()
uuid=uuid.decode() uuid = uuid.decode()
data = self.deserialize(data)
instance = factory.construct_from_net(str_type)(owner=owner, uuid=uuid) instance = factory.construct_from_net(str_type)(owner=owner, uuid=uuid, buffer=data)
# instance.data = instance.deserialize(data) # instance.data = instance.deserialize(data)
return instance return instance
def store(self, dict, persistent=False): def store(self, dict, persistent=False):
""" """
I want to store my replicated data. Persistent means into the disk I want to store my replicated data. Persistent means into the disk
If uuid is none we delete the key from the volume If uuid is none we delete the key from the volume
""" """
if self.uuid is not None: if self.uuid is not None:
if self.data == 'None': if self.buffer == 'None':
logger.info("erasing key {}".format(self.uuid)) logger.debug("erasing key {}".format(self.uuid))
del dict[self.uuid] del dict[self.uuid]
else: else:
dict[self.uuid] = self dict[self.uuid] = self
return self.uuid return self.uuid
def deserialize(self,data): def deserialize(self, data):
""" """
I want to apply changes into the DCC BUFFER -> JSON
MUST RETURN AN OBJECT INSTANCE
""" """
raise NotImplementedError raise NotImplementedError
def serialize(self,data): def serialize(self, data):
""" """
I want to load data from DCC I want to load data from DCC
DCC -> JSON
MUST RETURN A BYTE ARRAY MUST RETURN A BYTE ARRAY
""" """
raise NotImplementedError raise NotImplementedError
def apply(self,data,target):
"""
JSON -> DCC
"""
raise NotImplementedError
def resolve(self):
"""
I want to resolve my orphan data to an existing one
= Assing my pointer
"""
raise NotImplementedError
def dump(self):
return self.deserialize(self.buffer)
class RepCommand(ReplicatedDatablock):
def serialize(self,data):
return pickle.dumps(data)
def deserialize(self,data):
return pickle.load(data)
def apply(self,data,target):
target = data
# class RepObject(ReplicatedDatablock): # class RepObject(ReplicatedDatablock):
# def deserialize(self): # def deserialize(self):
# try: # try:
@ -179,5 +221,3 @@ class ReplicatedDatablock(object):
# def deserialize(self): # def deserialize(self):
# self.data = dump_datablock(self.pointer, 1) # self.data = dump_datablock(self.pointer, 1)

135
replication_client.py
View File

@ -2,10 +2,10 @@ import threading
import logging import logging
import zmq import zmq
import time import time
from replication import ReplicatedDatablock from replication import ReplicatedDatablock, RepCommand
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__) logger = logging.getLogger(__name__)
STATE_INITIAL = 0 STATE_INITIAL = 0
STATE_SYNCING = 1 STATE_SYNCING = 1
@ -14,16 +14,22 @@ STATE_ACTIVE = 2
class Client(object): class Client(object):
def __init__(self,factory=None, config=None): def __init__(self,factory=None, config=None):
self._rep_store = {}
self._net = ClientNetService(self._rep_store)
assert(factory) assert(factory)
self._rep_store = {}
self._net_client = ClientNetService(
store_reference=self._rep_store,
factory=factory)
self._factory = factory self._factory = factory
def connect(self): def connect(self):
self._net.start() self._net_client.start()
def disconnect(self): def disconnect(self):
self._net.stop() self._net_client.stop()
def state(self):
return self._net_client.state
def register(self, object): def register(self, object):
""" """
@ -34,12 +40,13 @@ class Client(object):
new_item = self._factory.construct_from_dcc(object)(owner="client", data=object) new_item = self._factory.construct_from_dcc(object)(owner="client", data=object)
if new_item: if new_item:
log.info("Registering {} on {}".format(object,new_item.uuid)) logger.info("Registering {} on {}".format(object,new_item.uuid))
new_item.store(self._rep_store) new_item.store(self._rep_store)
log.info("Pushing changes...") logger.info("Pushing changes...")
new_item.push(self._net.publish) new_item.push(self._net_client.publish)
return new_item.uuid return new_item.uuid
else: else:
raise TypeError("Type not supported") raise TypeError("Type not supported")
@ -50,20 +57,25 @@ class Client(object):
def unregister(self,object): def unregister(self,object):
pass pass
class ClientNetService(threading.Thread): class ClientNetService(threading.Thread):
def __init__(self,store_reference=None): def __init__(self,store_reference=None, factory=None):
# Threading # Threading
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.name = "ClientNetLink" self.name = "ClientNetLink"
self.daemon = True self.daemon = True
self.exit_event = threading.Event()
self._exit_event = threading.Event()
self._factory = factory
self._store_reference = store_reference
assert(self._factory)
# Networking # Networking
self.context = zmq.Context.instance() self.context = zmq.Context.instance()
self.snapshot = self.context.socket(zmq.DEALER) self.snapshot = self.context.socket(zmq.DEALER)
self.snapshot.setsockopt(zmq.IDENTITY, b'SERVER')
self.snapshot.connect("tcp://127.0.0.1:5560") self.snapshot.connect("tcp://127.0.0.1:5560")
self.subscriber = self.context.socket(zmq.SUB) self.subscriber = self.context.socket(zmq.SUB)
@ -78,25 +90,55 @@ class ClientNetService(threading.Thread):
def run(self): def run(self):
log.info("Client is listening") logger.info("Client is online")
poller = zmq.Poller() poller = zmq.Poller()
poller.register(self.snapshot, zmq.POLLIN) poller.register(self.snapshot, zmq.POLLIN)
poller.register(self.subscriber, zmq.POLLIN) poller.register(self.subscriber, zmq.POLLIN)
poller.register(self.publish, zmq.POLLOUT) poller.register(self.publish, zmq.POLLOUT)
self.state = 1 while not self._exit_event.is_set():
"""NET OUT
Given the net state we do something:
SYNCING : Ask for snapshots
ACTIVE : Do nothing
"""
if self.state == STATE_INITIAL:
self.snapshot.send(b"SNAPSHOT_REQUEST")
self.state = STATE_SYNCING
while not self.exit_event.is_set():
items = dict(poller.poll(10)) """NET IN
Given the net state we do something:
SYNCING : Ask for snapshots
ACTIVE : Do nothing
"""
items = dict(poller.poll(1))
if self.snapshot in items:
if self.state == STATE_SYNCING:
datablock = ReplicatedDatablock.pull(self.snapshot, self._factory)
if isinstance(datablock, RepCommand):
# We receive updates from the server !
if self.subscriber in items:
if self.state == STATE_ACTIVE:
logger.debug("Receiving changes from server")
datablock = ReplicatedDatablock.pull(self.subscriber, self._factory)
datablock.store(self._store_reference)
if not items: if not items:
log.error("No request ") logger.error("No request ")
time.sleep(1) time.sleep(1)
def setup(self,id="Client"):
pass
def stop(self): def stop(self):
self.exit_event.set() self._exit_event.set()
self.snapshot.close() self.snapshot.close()
self.subscriber.close() self.subscriber.close()
@ -110,7 +152,6 @@ class Server():
def __init__(self,config=None, factory=None): def __init__(self,config=None, factory=None):
self._rep_store = {} self._rep_store = {}
self._net = ServerNetService(store_reference=self._rep_store, factory=factory) self._net = ServerNetService(store_reference=self._rep_store, factory=factory)
# self.serve()
def serve(self): def serve(self):
self._net.start() self._net.start()
@ -128,7 +169,9 @@ class ServerNetService(threading.Thread):
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.name = "ServerNetLink" self.name = "ServerNetLink"
self.daemon = True self.daemon = True
self.exit_event = threading.Event() self._exit_event = threading.Event()
# Networking
self._rep_store = store_reference self._rep_store = store_reference
self.context = zmq.Context.instance() self.context = zmq.Context.instance()
@ -160,33 +203,41 @@ class ServerNetService(threading.Thread):
self.pull.bind("tcp://*:5562") self.pull.bind("tcp://*:5562")
except zmq.error.ZMQError: except zmq.error.ZMQError:
log.error("Address already in use, change net config") logger.error("Address already in use, change net config")
def run(self): def run(self):
log.info("Server is listening") logger.debug("Server is online")
poller = zmq.Poller() poller = zmq.Poller()
poller.register(self.snapshot, zmq.POLLIN) poller.register(self.snapshot, zmq.POLLIN)
poller.register(self.pull, zmq.POLLIN) poller.register(self.pull, zmq.POLLIN)
self.state = STATE_ACTIVE self.state = STATE_ACTIVE
while not self.exit_event.is_set(): while not self._exit_event.is_set():
# Non blocking poller # Non blocking poller
socks = dict(poller.poll(1000)) socks = dict(poller.poll())
# Snapshot system for late join (Server - Client) # Snapshot system for late join (Server - Client)
# if self.snapshot in socks: if self.snapshot in socks:
# msg = self.snapshot.recv_multipart(zmq.DONTWAIT) msg = self.snapshot.recv_multipart(zmq.DONTWAIT)
# identity = msg[0] identity = msg[0]
# request = msg[1] request = msg[1]
if request == b"SNAPSHOT_REQUEST":
pass
else:
logger.info("Bad snapshot request")
break
for key, item in self._rep_store:
self.snapshot.send(identity, zmq.SNDMORE)
item.push(self.snapshot)
self.snapshot.send(identity, zmq.SNDMORE)
RepCommand(owner='server',data='SNAPSHOT_END').push(self.snapshot)
# if request == b"SNAPSHOT_REQUEST":
# pass
# else:
# logger.info("Bad snapshot request")
# break
# ordered_props = [(SUPPORTED_TYPES.index(k.split('/')[0]),k,v) for k, v in self.property_map.items()] # ordered_props = [(SUPPORTED_TYPES.index(k.split('/')[0]),k,v) for k, v in self.property_map.items()]
# ordered_props.sort(key=itemgetter(0)) # ordered_props.sort(key=itemgetter(0))
@ -204,19 +255,17 @@ class ServerNetService(threading.Thread):
# Regular update routing (Clients / Client) # Regular update routing (Clients / Client)
if self.pull in socks: if self.pull in socks:
log.info("Receiving changes from client") logger.debug("Receiving changes from client")
msg = ReplicatedDatablock.pull(self.pull, self.factory) datablock = ReplicatedDatablock.pull(self.pull, self.factory)
msg.store(self._rep_store) datablock.store(self._rep_store)
# msg = message.Message.recv(self.collector_sock)
# # logger.info("received object") # Update all clients
# # Update all clients datablock.push(self.publisher)
# msg.store(self.store)
# msg.send(self.pub_sock)
def stop(self): def stop(self):
self.exit_event.set() self._exit_event.set()
self.snapshot.close() self.snapshot.close()
self.pull.close() self.pull.close()

42
test_replication.py
View File

@ -30,7 +30,7 @@ class RepSampleData(ReplicatedDatablock):
class TestDataReplication(unittest.TestCase): class TestDataReplication(unittest.TestCase):
def test_setup_data_factory(self): def test_data_factory(self):
factory = ReplicatedDataFactory() factory = ReplicatedDataFactory()
factory.register_type(SampleData, RepSampleData) factory.register_type(SampleData, RepSampleData)
data_sample = SampleData() data_sample = SampleData()
@ -38,23 +38,41 @@ class TestDataReplication(unittest.TestCase):
self.assertEqual(isinstance(rep_sample,RepSampleData), True) self.assertEqual(isinstance(rep_sample,RepSampleData), True)
def test_replicate_client_data(self): def test_basic_client_start(self):
factory = ReplicatedDataFactory() factory = ReplicatedDataFactory()
factory.register_type(SampleData, RepSampleData) factory.register_type(SampleData, RepSampleData)
server_api = Server(factory=factory) server = Server(factory=factory)
server_api.serve() server.serve()
client_api = Client(factory=factory)
client_api.connect()
data_sample = SampleData() client = Client(factory=factory)
data_sample_key = client_api.register(data_sample) client.connect()
#Waiting for server to receive the datas time.sleep(1)
time.sleep(.1)
#Check if if receive them self.assertEqual(client.state(), 2)
self.assertNotEqual(server_api._rep_store[data_sample_key],None)
# def test_register_client_data(self):
# # Setup data factory
# factory = ReplicatedDataFactory()
# factory.register_type(SampleData, RepSampleData)
# server = Server(factory=factory)
# server.serve()
# client = Client(factory=factory)
# client.connect()
# client2 = Client(factory=factory)
# client2.connect()
# data_sample_key = client.register(SampleData())
# #Waiting for server to receive the datas
# time.sleep(1)
# #Check if the server receive them
# self.assertNotEqual(client2._rep_store[data_sample_key],None)