<filename>coordinator.py
# coding: utf-8

import csv
import logging
import argparse
import asyncio
import queue
import time
import json
import signal

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M:%S')
logger = logging.getLogger('coordinator')

MAX_N_BYTES = 16
CHUNK = 1024

# connectionsMap = {}
# queue_out = queue.Queue()

class Coordinator():
    def __init__(self, id_c, file_path, blob_size, file_out):
        self.id = id_c
        self.datastore = []
        self.connectionsMap = {}
        self.data_array = queue.Queue()  # array that stores results from mapping and reducing
        self.lost_msgs = queue.Queue()
        self.last_reduced = False
        self.start_time = None
        self.file_path = file_path
        self.file_read = False
        self.file_written = False
        self.blob_size = blob_size
        self.file_out = file_out
        self.i_am_main = False
        self.pending_coordinator_request = False
        self.clients_list = []

    def full_state(self):
        # logger.debug('CONNS MAP TOP SEND: %s', self.connectionsMap)
        return { 'datastore' : self.datastore[:], 
                 'connectionsMap' : self.connectionsMap.copy(), 
                 'data_array' : list(self.data_array.queue), 
                 'lost_msgs' : list(self.lost_msgs.queue),
                 'clients_list' : self.clients_list[:],
                 'last_reduced' : self.last_reduced,
                 'file_read' : self.file_read,
                 'file_written' : self.file_written,
                 'start_time' : self.start_time
                 }

    def restore_state(self, state):
        logger.debug('Restoring state')
        self.datastore = state['datastore']
        self.connectionsMap = state['connectionsMap']
        self.clients_list = state['clients_list']

        self.data_array.queue.clear()
        [self.data_array.put(i) for i in state['data_array']]

        self.lost_msgs.queue.clear()
        [self.lost_msgs.put(i) for i in state['lost_msgs']]
        for addr, msg in self.connectionsMap.items(): # treat last messages sent as lost messages
            self.lost_msgs.put(msg)

        # logger.debug('LOST MSGS RCVD: %s', list(self.lost_msgs.queue))

        self.last_reduced = state['last_reduced']
        self.file_read = state['file_read']
        self.file_written = state['file_written']
        self.start_time = state['start_time']

    def read_file(self):
        # load txt file and divide it into blobs
        with self.file_path as f:
            while True:
                blob = f.read(self.blob_size)
                if not blob:
                    break
                # This loop is used to not break word in half
                while not str.isspace(blob[-1]):
                    ch = f.read(1)
                    if not ch:
                        break
                    blob += ch
                logger.debug('\nBlob: %s', blob)
                self.datastore.append(blob)

        self.file_read = True
        logger.debug('Number of blobs: %s', len(self.datastore))

    def print_to_file(self):
        hist = self.data_array.get()
        # store final histogram into a CSV file
        with self.file_out as f:
            csv_writer = csv.writer(f, delimiter=',',
                                    quotechar='"', quoting=csv.QUOTE_MINIMAL)

            for w, c in hist:
                csv_writer.writerow([w, c])
        self.file_written = True
        logger.debug('PRINT TO FILE DONE')

    def proccess_msg(self, message_json, addr=None):
        try:
            message = json.loads(message_json)
        except json.decoder.JSONDecodeError:
            logger.debug('Error on JSON parse: %s', message_json[:30])
            return

        if(message['task'] == 'register'):
            if self.start_time is None:
                self.start_time = time.time()
            if addr is not None:
                self.clients_list.append(addr)
            return
        elif message['task'] == 'attempt_main':
            if message['value'] == self.id: # i am sending this to myself
                self.i_am_main = True
            self.pending_coordinator_request = True
            return
        elif message['task'] == 'map_reply' or message['task'] == 'reduce_reply':
            self.data_array.put(message['value'])
            return
        elif message['task'] == 'coordinator_reply':
            self.restore_state(message['value'])
            return
        else:
            # logger.debug('THIS IS NOT FOR ME !!! %s', message['task'])
            pass
        # return self.scheduler()

    def scheduler(self):
        if self.pending_coordinator_request:
            result = { 'task': 'coordinator_reply', 'value': self.full_state() }
            self.pending_coordinator_request = False
            return result

        #Tambem nao queremos fazer muito mais porque senao perdemos muito trabalho
        queueSize = self.data_array.qsize()
        lost_msgs_size = self.lost_msgs.qsize()
        if(len(self.datastore) > 0): # blobs available
            new_message = self.datastore.pop() # get blob from out queue
            result = {'task': 'map_request', 'value': new_message}
            return result
        elif(lost_msgs_size > 0):
            # logger.debug('LOST MSGS: %s', lost_msgs_size)
            result = self.lost_msgs.get()
            return result
        elif(queueSize >= 2):
            if(queueSize % 2 == 0): #enviamos 2 sempre que pares
                new_message = []
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                return result
            else: #enviamos 3 sempre que temos impares
                new_message = []
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                return result
        else: # done
            if not self.last_reduced and queueSize > 0:
                new_message = []
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                self.last_reduced = True
                return result
            if self.start_time is not None: # there are clients connected
                end = time.time()
                logger.info('TIME TAKEN: %f (s)', end-self.start_time)
                result = {'task': 'done', 'value': 'done'}
                return result


    def parse_msg(self, msg):
        msg_len = len(msg)
        return '0'*(MAX_N_BYTES-len(str(int(msg_len)))) + str(msg_len) + msg

    async def register(self, host, port, loop):
        try:
            reader, writer = await asyncio.open_connection(host, port, loop=loop) # open connection
        # except (ConnectionError, ConnectionRefusedError, TimeoutError):
        # except Exception:
        except:
            self.i_am_main = True
            return

        msg = { 'task' : 'attempt_main', 'value' : self.id }
        msg_json = json.dumps(msg)        
        parsed_msg = self.parse_msg(msg_json)

        while True:
            if self.i_am_main: # i am already the main coordinator
                # logger.debug('BREAK1')
                break

            # send request to main coordinator
            writer.write(parsed_msg.encode())
            await writer.drain()

            # receive data
            try: 
                data = await reader.read(MAX_N_BYTES)
            except ConnectionResetError:
                # if not data:
                # logger.debug('BREAK2')
                break # become main coordinator

            if not data:
                # logger.debug('BREAK2.1')
                break # become main coordinator

            addr = writer.get_extra_info('peername')
            
            logger.debug('Size %s', data.decode())

            cur_size = 0
            total_size = int(data.decode())
            final_str = ''

            while (total_size - cur_size) >= CHUNK :
                data = await reader.read(CHUNK)
                if not data:
                    # logger.debug('BREAK3')
                    break # become main coordinator

                final_str = final_str + data.decode()
                cur_size += len(data)

            if cur_size != total_size:
                data = await reader.read(total_size - cur_size)

            if not data:
                # logger.debug('BREAK4')
                break # become main coordinator

            final_str = final_str + data.decode()

            # logger.info('Received: %r ' % final_str )
            logger.info('(REGISTER) Received from: %s ', addr )
            # logger.info('(REGISTER) Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize())

            message = final_str
            self.proccess_msg(message)

            await asyncio.sleep(1)

        self.i_am_main = True
        return

    async def handle_client(self, reader, writer):

        while True:
            data = await reader.read(MAX_N_BYTES)
            addr = writer.get_extra_info('peername')
            addr_str = str(addr)
            
            if not data or data == '':
                logger.debug("THE MONKEY KILLED: %s", addr_str)
                lostMsg = self.connectionsMap.get(addr_str)
                # logger.debug("LOST MESSAGE: %s", lostMsg)

                if lostMsg != None and addr_str in self.clients_list: # killed was a client
                    self.lost_msgs.put(lostMsg)
                    self.clients_list.remove(addr_str)

                if len(self.clients_list) > 0:
                    break

            cur_size = 0
            total_size = int(data.decode())
            final_str = ''

            while (total_size - cur_size) >= CHUNK :
                data = await reader.read(CHUNK)
                final_str = final_str + data.decode()
                cur_size += len(data)

            # if cur_size != total_size:
            #     data = await reader.read(total_size - cur_size)

            data = await reader.read(total_size - cur_size)
            final_str = final_str + data.decode()

            # logger.info('Received: %r ' % final_str )
            logger.info('Received from: %s ', addr_str )

            message = final_str
            self.proccess_msg(message, addr_str)

            to_send = self.scheduler()
            # logger.debug(to_send)
            if to_send is not None:
                if addr_str in self.clients_list:
                    if self.connectionsMap.get(addr_str) != None:
                        self.connectionsMap.pop(addr_str)
                    self.connectionsMap[addr_str] = to_send

                msg_json = json.dumps(to_send)
                parsed_msg = self.parse_msg(msg_json)

                # logger.debug("Sending: %r " % parsed_msg)
                logger.info('Sending to: %s ', addr_str )
                # logger.info('Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize())

                # logger.info('CONNS MAP: %r' % self.connectionsMap)
                
                writer.write(parsed_msg.encode())
                await writer.drain()

        logger.debug("Close the client socket")
        writer.close()

def close_server(loop, server):
    loop.stop()
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

def main(args):

    coordinator = Coordinator(args.coordinator_id, args.file, args.blob_size, args.out)

    # register
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coordinator.register('localhost', args.port, loop))
    loop.close()

    loop = asyncio.new_event_loop()
    coro = asyncio.start_server(coordinator.handle_client, 'localhost', args.port, loop=loop)
    server = loop.run_until_complete(coro)

    if not coordinator.file_read:
        coordinator.read_file()

    # Serve requests until Ctrl+C is pressed
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

    close_server(loop, server)
    # coordinator.print_to_file()
    if not coordinator.file_written:
        coordinator.print_to_file()
        coordinator.file_written = True

    # # Close the server
    # server.close()
    # loop.run_until_complete(server.wait_closed())
    # loop.close()

    # print("data_array size: ", data_array.qsize())
    # hist = data_array.get()
    # # store final histogram into a CSV file
    # with args.out as f:
    #     csv_writer = csv.writer(f, delimiter=',',
    #                             quotechar='"', quoting=csv.QUOTE_MINIMAL)

    #     for w, c in hist:
    #         csv_writer.writerow([w, c])


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='MapReduce Coordinator')
    parser.add_argument('-p', dest='port', type=int,
                        help='coordinator port', default=8765)
    parser.add_argument('-f', dest='file', type=argparse.FileType('r',
                            encoding='UTF-8'), help='input file path')
    parser.add_argument('-o', dest='out', type=argparse.FileType('w',
                        encoding='UTF-8'), help='output file path', default='output.csv')
    parser.add_argument('-b', dest='blob_size', type=int,
                        help='blob size', default=1024)
    parser.add_argument('-i', dest='coordinator_id', type=int,
                        help='coordinator id', default=1)
    args = parser.parse_args()


main(args)

<filename>coordinator.py
# coding: utf-8

import csv
import logging
import argparse
import asyncio
import queue
import time
import json
import signal

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M:%S')
logger = logging.getLogger('coordinator')

MAX_N_BYTES = 16
CHUNK = 1024

# connectionsMap = {}
# queue_out = queue.Queue()

class Coordinator():
    def __init__(self, id_c, file_path, blob_size, file_out):
        self.id = id_c
        self.datastore = []
        self.connectionsMap = {}
        self.data_array = queue.Queue()  # array that stores results from mapping and reducing
        self.lost_msgs = queue.Queue()
        self.last_reduced = False
        self.start_time = None
        self.file_path = file_path
        self.file_read = False
        self.file_written = False
        self.blob_size = blob_size
        self.file_out = file_out
        self.i_am_main = False
        self.pending_coordinator_request = False
        self.clients_list = []

    def full_state(self):
        # logger.debug('CONNS MAP TOP SEND: %s', self.connectionsMap)
        return { 'datastore' : self.datastore[:], 
                 'connectionsMap' : self.connectionsMap.copy(), 
                 'data_array' : list(self.data_array.queue), 
                 'lost_msgs' : list(self.lost_msgs.queue),
                 'clients_list' : self.clients_list[:],
                 'last_reduced' : self.last_reduced,
                 'file_read' : self.file_read,
                 'file_written' : self.file_written,
                 'start_time' : self.start_time
                 }

    def restore_state(self, state):
        logger.debug('Restoring state')
        self.datastore = state['datastore']
        self.connectionsMap = state['connectionsMap']
        self.clients_list = state['clients_list']

        self.data_array.queue.clear()
        [self.data_array.put(i) for i in state['data_array']]

        self.lost_msgs.queue.clear()
        [self.lost_msgs.put(i) for i in state['lost_msgs']]
        for addr, msg in self.connectionsMap.items(): # treat last messages sent as lost messages
            self.lost_msgs.put(msg)

        # logger.debug('LOST MSGS RCVD: %s', list(self.lost_msgs.queue))

        self.last_reduced = state['last_reduced']
        self.file_read = state['file_read']
        self.file_written = state['file_written']
        self.start_time = state['start_time']

    def read_file(self):
        # load txt file and divide it into blobs
        with self.file_path as f:
            while True:
                blob = f.read(self.blob_size)
                if not blob:
                    break
                # This loop is used to not break word in half
                while not str.isspace(blob[-1]):
                    ch = f.read(1)
                    if not ch:
                        break
                    blob += ch
                logger.debug('\nBlob: %s', blob)
                self.datastore.append(blob)

        self.file_read = True
        logger.debug('Number of blobs: %s', len(self.datastore))

    def print_to_file(self):
        hist = self.data_array.get()
        # store final histogram into a CSV file
        with self.file_out as f:
            csv_writer = csv.writer(f, delimiter=',',
                                    quotechar='"', quoting=csv.QUOTE_MINIMAL)

            for w, c in hist:
                csv_writer.writerow([w, c])
        self.file_written = True
        logger.debug('PRINT TO FILE DONE')

    def proccess_msg(self, message_json, addr=None):
        try:
            message = json.loads(message_json)
        except json.decoder.JSONDecodeError:
            logger.debug('Error on JSON parse: %s', message_json[:30])
            return

        if(message['task'] == 'register'):
            if self.start_time is None:
                self.start_time = time.time()
            if addr is not None:
                self.clients_list.append(addr)
            return
        elif message['task'] == 'attempt_main':
            if message['value'] == self.id: # i am sending this to myself
                self.i_am_main = True
            self.pending_coordinator_request = True
            return
        elif message['task'] == 'map_reply' or message['task'] == 'reduce_reply':
            self.data_array.put(message['value'])
            return
        elif message['task'] == 'coordinator_reply':
            self.restore_state(message['value'])
            return
        else:
            # logger.debug('THIS IS NOT FOR ME !!! %s', message['task'])
            pass
        # return self.scheduler()

    def scheduler(self):
        if self.pending_coordinator_request:
            result = { 'task': 'coordinator_reply', 'value': self.full_state() }
            self.pending_coordinator_request = False
            return result

        #Tambem nao queremos fazer muito mais porque senao perdemos muito trabalho
        queueSize = self.data_array.qsize()
        lost_msgs_size = self.lost_msgs.qsize()
        if(len(self.datastore) > 0): # blobs available
            new_message = self.datastore.pop() # get blob from out queue
            result = {'task': 'map_request', 'value': new_message}
            return result
        elif(lost_msgs_size > 0):
            # logger.debug('LOST MSGS: %s', lost_msgs_size)
            result = self.lost_msgs.get()
            return result
        elif(queueSize >= 2):
            if(queueSize % 2 == 0): #enviamos 2 sempre que pares
                new_message = []
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                return result
            else: #enviamos 3 sempre que temos impares
                new_message = []
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                return result
        else: # done
            if not self.last_reduced and queueSize > 0:
                new_message = []
                new_message.append(self.data_array.get())
                result = {'task': 'reduce_request', 'value': new_message}
                self.last_reduced = True
                return result
            if self.start_time is not None: # there are clients connected
                end = time.time()
                logger.info('TIME TAKEN: %f (s)', end-self.start_time)
                result = {'task': 'done', 'value': 'done'}
                return result


    def parse_msg(self, msg):
        msg_len = len(msg)
        return '0'*(MAX_N_BYTES-len(str(int(msg_len)))) + str(msg_len) + msg

    async def register(self, host, port, loop):
        try:
            reader, writer = await asyncio.open_connection(host, port, loop=loop) # open connection
        # except (ConnectionError, ConnectionRefusedError, TimeoutError):
        # except Exception:
        except:
            self.i_am_main = True
            return

        msg = { 'task' : 'attempt_main', 'value' : self.id }
        msg_json = json.dumps(msg)        
        parsed_msg = self.parse_msg(msg_json)

        while True:
            if self.i_am_main: # i am already the main coordinator
                # logger.debug('BREAK1')
                break

            # send request to main coordinator
            writer.write(parsed_msg.encode())
            await writer.drain()

            # receive data
            try: 
                data = await reader.read(MAX_N_BYTES)
            except ConnectionResetError:
                # if not data:
                # logger.debug('BREAK2')
                break # become main coordinator

            if not data:
                # logger.debug('BREAK2.1')
                break # become main coordinator

            addr = writer.get_extra_info('peername')
            
            logger.debug('Size %s', data.decode())

            cur_size = 0
            total_size = int(data.decode())
            final_str = ''

            while (total_size - cur_size) >= CHUNK :
                data = await reader.read(CHUNK)
                if not data:
                    # logger.debug('BREAK3')
                    break # become main coordinator

                final_str = final_str + data.decode()
                cur_size += len(data)

            if cur_size != total_size:
                data = await reader.read(total_size - cur_size)

            if not data:
                # logger.debug('BREAK4')
                break # become main coordinator

            final_str = final_str + data.decode()

            # logger.info('Received: %r ' % final_str )
            logger.info('(REGISTER) Received from: %s ', addr )
            # logger.info('(REGISTER) Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize())

            message = final_str
            self.proccess_msg(message)

            await asyncio.sleep(1)

        self.i_am_main = True
        return

    async def handle_client(self, reader, writer):

        while True:
            data = await reader.read(MAX_N_BYTES)
            addr = writer.get_extra_info('peername')
            addr_str = str(addr)
            
            if not data or data == '':
                logger.debug("THE MONKEY KILLED: %s", addr_str)
                lostMsg = self.connectionsMap.get(addr_str)
                # logger.debug("LOST MESSAGE: %s", lostMsg)

                if lostMsg != None and addr_str in self.clients_list: # killed was a client
                    self.lost_msgs.put(lostMsg)
                    self.clients_list.remove(addr_str)

                if len(self.clients_list) > 0:
                    break

            cur_size = 0
            total_size = int(data.decode())
            final_str = ''

            while (total_size - cur_size) >= CHUNK :
                data = await reader.read(CHUNK)
                final_str = final_str + data.decode()
                cur_size += len(data)

            # if cur_size != total_size:
            #     data = await reader.read(total_size - cur_size)

            data = await reader.read(total_size - cur_size)
            final_str = final_str + data.decode()

            # logger.info('Received: %r ' % final_str )
            logger.info('Received from: %s ', addr_str )

            message = final_str
            self.proccess_msg(message, addr_str)

            to_send = self.scheduler()
            # logger.debug(to_send)
            if to_send is not None:
                if addr_str in self.clients_list:
                    if self.connectionsMap.get(addr_str) != None:
                        self.connectionsMap.pop(addr_str)
                    self.connectionsMap[addr_str] = to_send

                msg_json = json.dumps(to_send)
                parsed_msg = self.parse_msg(msg_json)

                # logger.debug("Sending: %r " % parsed_msg)
                logger.info('Sending to: %s ', addr_str )
                # logger.info('Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize())

                # logger.info('CONNS MAP: %r' % self.connectionsMap)
                
                writer.write(parsed_msg.encode())
                await writer.drain()

        logger.debug("Close the client socket")
        writer.close()

def close_server(loop, server):
    loop.stop()
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

def main(args):

    coordinator = Coordinator(args.coordinator_id, args.file, args.blob_size, args.out)

    # register
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coordinator.register('localhost', args.port, loop))
    loop.close()

    loop = asyncio.new_event_loop()
    coro = asyncio.start_server(coordinator.handle_client, 'localhost', args.port, loop=loop)
    server = loop.run_until_complete(coro)

    if not coordinator.file_read:
        coordinator.read_file()

    # Serve requests until Ctrl+C is pressed
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

    close_server(loop, server)
    # coordinator.print_to_file()
    if not coordinator.file_written:
        coordinator.print_to_file()
        coordinator.file_written = True

    # # Close the server
    # server.close()
    # loop.run_until_complete(server.wait_closed())
    # loop.close()

    # print("data_array size: ", data_array.qsize())
    # hist = data_array.get()
    # # store final histogram into a CSV file
    # with args.out as f:
    #     csv_writer = csv.writer(f, delimiter=',',
    #                             quotechar='"', quoting=csv.QUOTE_MINIMAL)

    #     for w, c in hist:
    #         csv_writer.writerow([w, c])


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='MapReduce Coordinator')
    parser.add_argument('-p', dest='port', type=int,
                        help='coordinator port', default=8765)
    parser.add_argument('-f', dest='file', type=argparse.FileType('r',
                            encoding='UTF-8'), help='input file path')
    parser.add_argument('-o', dest='out', type=argparse.FileType('w',
                        encoding='UTF-8'), help='output file path', default='output.csv')
    parser.add_argument('-b', dest='blob_size', type=int,
                        help='blob size', default=1024)
    parser.add_argument('-i', dest='coordinator_id', type=int,
                        help='coordinator id', default=1)
    args = parser.parse_args()


main(args)

# coding: utf-8 # connectionsMap = {} # queue_out = queue.Queue() # array that stores results from mapping and reducing # logger.debug('CONNS MAP TOP SEND: %s', self.connectionsMap) # treat last messages sent as lost messages # logger.debug('LOST MSGS RCVD: %s', list(self.lost_msgs.queue)) # load txt file and divide it into blobs # This loop is used to not break word in half # store final histogram into a CSV file # i am sending this to myself # logger.debug('THIS IS NOT FOR ME !!! %s', message['task']) # return self.scheduler() #Tambem nao queremos fazer muito mais porque senao perdemos muito trabalho # blobs available # get blob from out queue # logger.debug('LOST MSGS: %s', lost_msgs_size) #enviamos 2 sempre que pares #enviamos 3 sempre que temos impares # done # there are clients connected # open connection # except (ConnectionError, ConnectionRefusedError, TimeoutError): # except Exception: # i am already the main coordinator # logger.debug('BREAK1') # send request to main coordinator # receive data # if not data: # logger.debug('BREAK2') # become main coordinator # logger.debug('BREAK2.1') # become main coordinator # logger.debug('BREAK3') # become main coordinator # logger.debug('BREAK4') # become main coordinator # logger.info('Received: %r ' % final_str ) # logger.info('(REGISTER) Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize()) # logger.debug("LOST MESSAGE: %s", lostMsg) # killed was a client # if cur_size != total_size: #     data = await reader.read(total_size - cur_size) # logger.info('Received: %r ' % final_str ) # logger.debug(to_send) # logger.debug("Sending: %r " % parsed_msg) # logger.info('Datastore: %s, Data_Array: %s', len(self.datastore), self.data_array.qsize()) # logger.info('CONNS MAP: %r' % self.connectionsMap) # register # Serve requests until Ctrl+C is pressed # coordinator.print_to_file() # # Close the server # server.close() # loop.run_until_complete(server.wait_closed()) # loop.close() # print("data_array size: ", data_array.qsize()) # hist = data_array.get() # # store final histogram into a CSV file # with args.out as f: #     csv_writer = csv.writer(f, delimiter=',', #                             quotechar='"', quoting=csv.QUOTE_MINIMAL) #     for w, c in hist: #         csv_writer.writerow([w, c])