research/data_sync/sync.py

# Sync protocol PoC

import hashlib
import networksim
import networkwhisper
import random
import sync_pb2
import time

# Each group belongs to a client.
# Hardcoded for now.
# group\_id = HASH("GROUP\_ID", client\_id, group\_descriptor)
GROUP_ID = "0xdeadbeef"

# TODO: Introduce exponential back-off for send_time based on send_count

# XXX: Add debug log level
def log(message):
    print(message)

def merge_two_dicts(x, y):
    z = x.copy()
    z.update(y)
    return z

class Node():
    def __init__(self, name, network, profile, mode='batch'):
        self.name = name
        self.log = []
        self.messages = {}
        self.sync_state = {}
        self.peers = {}
        self.network = network
        self.time = 0
        self.mode = mode
        self.offeredMessages = {} # XXX: Should be bounded

        # XXX: Assumes only one group
        self.group_id = GROUP_ID
        self.sharing = {GROUP_ID: set()}

        self.profile = profile
        # for index in pulsating reseries if mobile node

        # XXX: Hacky
        if (self.name == 'A'):
            self.randomSeed = 0
        elif (self.name == 'B'):
            self.randomSeed = 1
        else:
            self.randomSeed = random.randint(1,10)

        if profile == 'burstyMobile':
            self.reliability = 0.1
            self.update_availability()
        elif profile == 'onlineDesktop':
            self.reliability = 1 # or 0.9
        else:
            self.reliability = 1
        self.availability = self.reliability

    def tick(self):
        # XXX: What else do?
        # TODO: Send message if reached send time
        self.time += 1

        if (self.profile == 'burstyMobile'):
            self.update_availability()

        if (self.availability == 1):
            if (self.mode == 'batch'):
                self.ack_sent_messages()
                self.ack_offered_messages()
                self.req_offered_messages()
                self.send_requested_messages()
                self.send_messages()
            elif (self.mode == 'interactive'):
                self.ack_received_messages()
                self.ack_offered_messages()
                self.req_offered_messages()
                self.send_requested_messages()
                self.offer_messages()
        #elif (self.availability == 0):
            #print "*** node NOT available", self.name
        #else:
        #    print "*** conflation overload, reliability/availability mismatch"

    # NOTE: Assuming same semantics for batch and interactive mode.
    #- **Acknowledge** any messages **received** from the peer that the device has
    #not yet acknowledged
    def ack_received_messages(self):
        self.ack_sent_messages()

    # - **Acknowledge** any messages **sent** by the peer that the device has not yet
    #   acknowledged
    def ack_sent_messages(self):
        # TODO: Accumulate and send all in one go

        # XXX: Better Pythonesque way to do this
        for mid, x in self.sync_state.items():
            for peer, flags in x.items():
                if flags['ack_flag'] == 1:
                    ack_rec = new_ack_record([mid])
                    self.network.send_message(self.name, peer, ack_rec)
                    self.sync_state[mid][peer]['ack_flag'] = 0
                    log("    ACK ({} -> {}): {}".format(self.name, peer, mid[:4]))

    # - **Acknowledge** any messages **offered** by the peer that the device holds,
    #   and has not yet acknowledged
    # ACK maybe once?!
    def ack_offered_messages(self):
        for peer, message_ids in self.offeredMessages.items():
            for message_id in message_ids:
                if (message_id in self.messages and
                    # XXX: What if they didn't receive ACK?
                    self.sync_state[message_id][peer]['ack_flag'] == 1):
                    # XXX: Slurp up
                    ack_rec = new_ack_record([message_id])
                    self.sync_state[message_id][peer]['ack_flag'] = 0
                    self.network.send_message(self.name, peer, ack_rec)

    #  **Request** any messages **offered** by the peer that the device does not
    #   hold, and has not yet requested
    # NOTE: (Overloaded?) use of send_time and send_count for reqs.
    # Seems OK since hold flag clarifies if you need to offer/send or ack.
    def req_offered_messages(self):
        # XXX: Not removing from cache, instead letting it grow indefinitely
        # (later: bounded) UNLESS ACK etc is received
        for peer_id, message_ids in self.offeredMessages.items():
            for message_id in message_ids:
                if (message_id not in self.messages and
                    # XXX: Not clear this is part of spec
                    self.sync_state[message_id][peer_id]['send_time'] <= self.time
                    ):
                    # XXX: Slurp up
                    req_rec = new_req_record([message_id])
                    self.network.send_message(self.name, peer_id, req_rec)

                    n = self.sync_state[message_id][peer_id]["send_count"] + 1
                    self.update_sync_state(message_id, peer_id, {
                        'hold_flag': 1,
                        'send_count': n,
                        'send_time': self.time + int(n**2) + 1
                    })

                    log("REQUEST ({} -> {}): {}".format(self.name, peer_id, message_id[:4]))
                    # XXX: It is double requesting, should be polite

    # - **Send** any messages that the device is **sharing** with the peer, that have
    #   been **requested** by the peer, and that have reached their send times
    def send_requested_messages(self):
        for message_id, x in self.sync_state.items():
            for peer_id, flags in x.items():
                if (peer_id in self.sharing[self.group_id] and
                    flags['request_flag'] == 1 and
                    flags['send_time'] <= self.time):
                    message = self.messages[message_id]
                    send_count = self.sync_state[message_id][peer_id]["send_count"] + 1
                    self.sync_state[message_id][peer_id]["send_count"] = send_count
                    self.sync_state[message_id][peer_id]["send_time"] += self.time + send_count*2
                    self.sync_state[message_id][peer_id]["request_flag"] = 0
                    log('MESSAGE ({} -> {}): {} requested and sent'.format(self.name, peer_id, message_id [:4]))
                    # XXX: Can introduce latency here
                    self.network.send_message(self.name, peer_id, message)

    # When turn off request flag?

    #- **Offer** any messages that the device is **sharing** with the peer, and does
    #  not know whether the peer holds, and that have reached their send times
    # XXX: Not tested yet, interactive mode
    def offer_messages(self):
        for message_id, x in self.sync_state.items():
            for peer_id, flags in x.items():
                ids = []
                if (peer_id in self.sharing[self.group_id] and
                    flags['hold_flag'] == 0 and
                    flags['send_time'] <= self.time):
                    # TODO: Extend to slurp up all, need index peer->message
                    offer_rec = new_offer_record([message_id])
                    self.network.send_message(self.name, peer_id, offer_rec)
                    send_count = self.sync_state[message_id][peer_id]["send_count"] + 1
                    self.sync_state[message_id][peer_id]["send_count"] = send_count
                    self.sync_state[message_id][peer_id]["send_time"] += self.time + send_count*2
                    log("  OFFER ({} -> {}): {}".format(self.name, peer_id, message_id[:4]))

    # - **Send** any messages that the device is **sharing** with the peer, and does
    #   not know whether the peer holds, and that have reached their send times
    def send_messages(self):
        for message_id, x in self.sync_state.items():
            for peer_id, flags in x.items():
                # Should be case for B no?
                if (peer_id in self.sharing[self.group_id] and
                    flags['hold_flag'] == 0 and
                    flags['send_time'] <= self.time):
                    message = self.messages[message_id]
                    send_count = self.sync_state[message_id][peer_id]["send_count"] + 1
                    self.sync_state[message_id][peer_id]["send_count"] = send_count
                    self.sync_state[message_id][peer_id]["send_time"] += self.time + send_count*2
                    log('MESSAGE ({} -> {}): {} sent'.format(self.name, peer_id, message_id [:4]))
                    # XXX: Can introduce latency here
                    self.network.send_message(self.name, peer_id, message)

    # XXX: Why would node know about peer and not just name?
    def addPeer(self, peer_id, peer):
        self.peers[peer_id] = peer

    def share(self, peer_id):
        self.sharing[self.group_id].add(peer_id)

    # Helper method
    def update_sync_state(self, message_id, peer_id, new_state):
        if message_id not in self.sync_state:
            self.sync_state[message_id] = {}
        if peer_id not in self.sync_state[message_id]:
            self.sync_state[message_id][peer_id] = {
                "hold_flag": 0,
                "ack_flag": 0,
                "request_flag": 0,
                "send_count": 0,
                "send_time": self.time + 1
            }

        current = self.sync_state[message_id][peer_id]
        new = merge_two_dicts(current, new_state)
        self.sync_state[message_id][peer_id] = new

    def append_message(self, message):
        message_id = get_message_id(message)
        self.log.append({"id": message_id,
                         "message": message})
        # XXX: Ugly but easier access while keeping log order
        self.messages[message_id] = message
        self.sync_state[message_id] = {}
        # Ensure added for each peer
        # If we add peer at different time, ensure state init
        # TODO: Only share with certain peers, e.g. clientPolicy
        for peer in self.peers.keys():
            if peer in self.sharing[self.group_id]:
                self.sync_state[message_id][peer] = {
                    "hold_flag": 0,
                    "ack_flag": 0,
                    "request_flag": 0,
                    "send_count": 0,
                    "send_time": self.time + 1
                }

    def on_receive(self, sender, message):
        if random.random() < self.reliability:
            #print "*** {} received message from {}".format(self.name, sender.name)
            if (message.header.type == 1):
                self.on_receive_message(sender, message)
            elif (message.header.type == 0):
                self.on_receive_ack(sender, message)
            elif (message.header.type == 2):
                self.on_receive_offer(sender, message)
            elif (message.header.type == 3):
                self.on_receive_request(sender, message)
            else:
                print("XXX: unknown message type")
        else:
            log("*** node {} offline, dropping message".format(self.name))

    def on_receive_message(self, sender, message):
        message_id = get_message_id(message)
        log('MESSAGE ({} -> {}): {} received'.format(sender.name, self.name, message_id[:4]))
        if message_id not in self.sync_state:
            self.sync_state[message_id] = {}

        if sender.name in self.sync_state[message_id]:
            self.sync_state[message_id][sender.name]['hold_flag'] == 1
            self.sync_state[message_id][sender.name]['ack_flag'] == 1
            # XXX: ACK again here?
        self.sync_state[message_id][sender.name] = {
            "hold_flag": 1,
            "ack_flag": 1,
            "request_flag": 0,
            "send_count": 0,
            "send_time": 0
        }

        # XXX: If multiple group id, dispatch per group id
        for peer in self.sharing[self.group_id]:
            if peer not in self.sync_state[message_id]:
                self.sync_state[message_id][peer] = {
                    "hold_flag": 0,
                    "ack_flag": 0,
                    "request_flag": 0,
                    "send_count": 0,
                    "send_time": 0
                }

        self.messages[message_id] = message

    def on_receive_ack(self, sender, message):
        for ack in message.payload.ack.id:
            log('    ACK ({} -> {}): {} received'.format(sender.name, self.name, ack[:4]))
            self.sync_state[ack][sender.name]["hold_flag"] = 1

    def on_receive_offer(self, sender, message):
        for message_id in message.payload.offer.id:
            log('  OFFER ({} -> {}): {} received'.format(sender.name, self.name, message_id[:4]))
            if (message_id in self.sync_state and
                sender.name in self.sync_state[message_id] and
                self.sync_state[message_id][sender.name]['ack_flag'] == 1):
                print("Have message, not ACKED yet, add to list", sender.name, message_id)
                if sender.name not in self.offeredMessages:
                    self.offeredMessages[sender.name] = []
                self.offeredMessages[sender.name].append(message_id)
            elif message_id not in self.sync_state:
                #print "*** {} on_receive_offer from {} not holding {}".format(self.name, sender.name, message_id)
                if sender.name not in self.offeredMessages:
                    self.offeredMessages[sender.name] = []
                self.offeredMessages[sender.name].append(message_id)
            #else:
            #    print "*** {} on_receive_offer have {} and ACKd OR peer {} unknown".format(self.name, message_id, sender.name)

            # XXX: Init fn to wrap updates
            if message_id not in self.sync_state:
                self.sync_state[message_id] = {}
            if sender.name not in self.sync_state[message_id]:
                self.sync_state[message_id][sender.name] = {
                    "hold_flag": 1,
                    "ack_flag": 0,
                    "request_flag": 0,
                    "send_count": 0,
                    "send_time": 0
                }
            self.sync_state[message_id][sender.name]['hold_flag'] = 1
            #print "*** {} offeredMessages {}".format(self.name, self.offeredMessages)

    def on_receive_request(self, sender, message):
        for req in message.payload.request.id:
            log('REQUEST ({} -> {}): {} received'.format(sender.name, self.name, req[:4]))
            self.sync_state[req][sender.name]["request_flag"] = 1

    def print_sync_state(self):
        log("\n{} POV @{}".format(self.name, self.time))
        log("-" * 60)
        n = self.name
        for message_id, x in self.sync_state.items():
            line = message_id[:4] + " | "
            for peer, flags in x.items():
                line += peer + ": "
                if flags['hold_flag']:
                    line += "hold "
                if flags['ack_flag']:
                    line += "ack "
                if flags['request_flag']:
                    line += "req "
                line += "@" + str(flags['send_time'])
                line += "(" + str(flags['send_count']) + ")"
                line += " | "

            log(line)
        #log("-" * 60)

    # Shorter names for pubkey
    def print_sync_state2(self):
        log("\n{} POV @{}".format(self.name[-4:], self.time))
        log("-" * 60)
        n = self.name[-4:]
        for message_id, x in self.sync_state.items():
            line = message_id[:4] + " | "
            for peer, flags in x.items():
                line += peer[-4:] + ": "
                if flags['hold_flag']:
                    line += "hold "
                if flags['ack_flag']:
                    line += "ack "
                if flags['request_flag']:
                    line += "req "
                line += "@" + str(flags['send_time'])
                line += "(" + str(flags['send_count']) + ")"
                line += " | "

            log(line)
        #log("-" * 60)

    def update_availability(self):
        #arr = [1, 1, 1, 1, 1, 0, 0, 0, 0, 0]
        arr = [1, 1, 0, 0, 1, 1, 0, 0]
        idx = (self.time + self.randomSeed) % 8 # 10
        self.reliability = arr[idx]
        # XXX conflating these for now, depends on POV/agency
        self.availability = arr[idx]

# XXX: Self-describing better in practice, format?
def sha1(message):
    # XXX correct encoding?
    sha = hashlib.sha1(message.encode('utf-8'))
    return sha.hexdigest()

#- message\_id = HASH("MESSAGE\_ID", group\_id, timestamp, message\_body)
# TODO: Create a message
def create_message(body):
    group_id = "0xdeadbeef"
    timestamp = time.time()
    message_body = body
    message = {"group_id": group_id, "timestamp": timestamp, "message_body": message_body}
    return message

# XXX: Is this hashing correctly?
def get_message_id(message_record):
    msg = message_record.payload.message
    s = "MESSAGE_ID" + msg.group_id + str(msg.timestamp) + msg.body.decode()
    #print("***", s)
    return sha1(s)

# TODO: Move these protobuf helpers somewhere better

# XXX: where is the message id?
def new_message_record(body):
    msg = sync_pb2.Record()
    msg.header.version = 1
    # assert based on type and length
    msg.header.type = 1 # MESSAGE type
    # XXX: Should be inferred
    msg.header.length = 10
    # XXX: Hardcoded for now
    msg.payload.message.group_id = "foo"
    # XXX: Should be 64 bit integer ms
    msg.payload.message.timestamp = int(time.time())
    msg.payload.message.body = str.encode(body)
    return msg

def new_ack_record(ids):
    msg = sync_pb2.Record()
    msg.header.version = 1
    # assert based on type and length
    msg.header.type = 0 # ACK type
    # XXX: Should be inferred
    msg.header.length = 10
    for id in ids:
        msg.payload.ack.id.append(id)
    return msg

def new_offer_record(ids):
    msg = sync_pb2.Record()
    msg.header.version = 1
    # assert based on type and length
    msg.header.type = 2 # OFFER type
    # XXX: Should be inferred
    msg.header.length = 10
    for id in ids:
        msg.payload.offer.id.append(id)
    return msg

def new_req_record(ids):
    msg = sync_pb2.Record()
    msg.header.version = 1
    # assert based on type and length
    msg.header.type = 3 # REQUEST type
    # XXX: Should be inferred
    msg.header.length = 10
    for id in ids:
        msg.payload.request.id.append(id)
    return msg


# Mocking
################################################################################

# TODO: For whisper nodes should be public keys
# What about keypair to try to decrypt? should be in node

def run(steps=10):
    n = networksim.NetworkSimulator()

    # XXX: Not clear to me what's best here
    # Interactive: less BW, Batch: less coordination
    a = Node("A", n, 'burstyMobile', 'batch')
    b = Node("B", n, 'burstyMobile', 'batch')
    c = Node("C", n, 'desktop', 'interactive')
    d = Node("D", n, 'desktop', 'batch')

    n.peers["A"] = a
    n.peers["B"] = b
    n.peers["C"] = c
    n.peers["D"] = d
    n.nodes = [a, b, c, d]

    a.addPeer("B", b)
    a.addPeer("C", c)
    b.addPeer("A", a)
    c.addPeer("A", a)

    #b.addPeer("C", c) # hm
    #c.addPeer("B", b)

    b.addPeer("D", d)
    c.addPeer("D", d)

    # NOTE: Client should decide policy, implict group
    a.share("B")
    b.share("A")

    # C and D participating
    # a.share("C")
    b.share("D")
    c.share("A")
    c.share("D")
    d.share("B")
    d.share("C")

    print("\nAssuming one group context (A-B (C-D) share):")

    # XXX: Conditional append to get message graph?
    # TODO: Actually need to encode graph, client concern
    local_appends = {
        1: [[a, "A: hello world"]],
        2: [[b, "B: hello!"]],
    }

    for i in range(steps):
        # NOTE: include signature and parent message
        if n.time in local_appends:
            for peer, msg in local_appends[n.time]:
                rec = new_message_record(msg)
                peer.append_message(rec)

        n.tick()
        #a.print_sync_state()
        #b.print_sync_state()
        #c.print_sync_state()

    a.print_sync_state()
    b.print_sync_state()
    c.print_sync_state()
    d.print_sync_state()

def whisperRun(steps=10):
    a_keyPair = "0x57083392b29bdf24512c93cfdf45d38c87d9d882da3918c59f4406445ea976a4"
    b_keyPair= "0x7b5c5af9736d9f1773f2020dd0fef0bc3c8aeaf147d2bf41961e766588e086e7"

    # TODO: should be node names
    # Derived, used for addressing
    a_pubKey = "0x04d94a1a01872b598c7cdc5aca2358d35eb91cd8a91eaea8da277451bb71d45c0d1eb87a31ea04e32f537e90165c870b3e115a12438c754d507ac75bddd6ecacd5"
    b_pubKey = "0x04ff921ddf78b5ed4537402f59a150caf9d96a83f2a345a1ddf9df12e99e7778f314c9ca72e8285eb213af84f5a7b01aabb62c67e46657976ded6658e1b9e83c73"

    aNode = networkwhisper.WhisperNodeHelper(a_keyPair)
    bNode = networkwhisper.WhisperNodeHelper(b_keyPair)

    # XXX: Not clear to me what's best here
    # Interactive: less BW, Batch: less coordination
    a = Node(a_pubKey, aNode, 'burstyMobile', 'batch')
    b = Node(b_pubKey, bNode, 'burstyMobile', 'batch')

    # XXX: Not clear this is needed for Whisper, since all nodes should be part of network
    # Possibly analog with topics?
    #n.peers["A"] = a
    #n.peers["B"] = b
    aNode.nodes = [a]
    bNode.nodes = [b]

    a.addPeer(b_pubKey, b)
    b.addPeer(a_pubKey, a)

    # NOTE: Client should decide policy, implict group
    a.share(b_pubKey)
    b.share(a_pubKey)

    print("\nAssuming one group context (A-B) share):")

    # XXX: Conditional append to get message graph?
    # TODO: Actually need to encode graph, client concern
    local_appends = {
        1: [[a, "A: hello world"]],
        2: [[b, "B: hello!"]],
    }

    # XXX: what is this again? should be for both nodes
    for i in range(steps):
        # NOTE: include signature and parent message
        if aNode.time in local_appends:
            for peer, msg in local_appends[aNode.time]:
                rec = new_message_record(msg)
                peer.append_message(rec)

        # XXX: Why discrete time model here?
        aNode.tick()
        bNode.tick()
        #a.print_sync_state()
        #b.print_sync_state()

    a.print_sync_state2()
    b.print_sync_state2()

#run(30)
whisperRun(30)