2
\$\begingroup\$

Following is the main file in the project (executing tasks). It uses the mpi4py module, and multiple processes execute the same code. Please review the code and post your suggestions on making it more efficient and pythonic.

from mpi4py import MPI
from Peers import Peers
from SegmentHandler import SegmentHandler
from Communicator import Communicator
import Message
import Datastore
import Queue
import random
import Logging


class JobScheduler:

    def __init__(self, environment):
        self.MAX_BACKLOG = 5
        self.SegmentAssignProcId = 0
        self.environment = environment
        self.segmentHandler = SegmentHandler()
        self.peers = Peers(self.environment.totalProc)
        self.dataHandler = Datastore.Datastore()

    def isSegmentAssigner(self):
        return self.environment.procId == self.SegmentAssignProcId

    def runMicroDownload(self):
        if self.isSegmentAssigner():
            self.microDownload()

    def initLocalQueue(self):
        self.toBeAdvertised = Queue.Queue()
        self.requestQueue = Queue.Queue()
        self.downloadRequests = Queue.Queue()

    def handleAdvertisementQueue(self):
        if not self.toBeAdvertised.empty():
            adMessage = self.toBeAdvertised.get()
            segmentId = adMessage.messageId
            Logging.logProcessOp(processId=self.environment.procId,
                                 op="pop",
                                 depQueue=self.toBeAdvertised,
                                 message=adMessage)
            reqMessage = adMessage.getResponse()

            # broadcast
            for _ in range(self.environment.totalProc):
                if _ != self.SegmentAssignProcId:
                    reqMessage.receiver = _
                    self.environment.send(_, reqMessage)

    def handleDownloadRequestQueue(self):
        if not self.downloadRequests.empty():
            dwnldReqMessage = self.downloadRequests.get()
            dwnldReqMessage.download()
            Logging.logProcessOp(processId=self.environment.procId,
                                 op="pop",
                                 depQueue=self.downloadRequests,
                                 message=dwnldReqMessage)
            self.dataHandler.addSegment(dwnldReqMessage.messageId,
                                        dwnldReqMessage.content)
            self.dataHandler.store(dwnldReqMessage)
            self.toBeAdvertised.put(dwnldReqMessage)
            # get response and send it to the inititor

    def handleRequestQueue(self):
        if not self.requestQueue.empty():
            reqMessage = self.requestQueue.get()
            responseMsg = reqMessage.getResponse()
            Logging.logProcessOp(processId=self.environment.procId,
                                 op="pop",
                                 depQueue=self.requestQueue,
                                 message=reqMessage)
            self.environment.send(responseMsg.receiver, responseMsg)

    def runMicroNC(self):
        if self.environment.procId != self.SegmentAssignProcId:
            self.microNC()

    def stopMicroNC(self):

        if self.toBeAdvertised.empty() and self.requestQueue.empty() and \
           self.downloadRequests.empty() and \
           self.dataHandler.downlodedAll(4):
            return True

        return False

    def microNC(self):

        Logging.logProcessOp(processId=self.environment.procId,
                             op="startMicroNC")

        self.initLocalQueue()

        while not self.stopMicroNC():

            nonDetchoice = random.randrange(4)

            if nonDetchoice == 0:

                _message = self.environment.nonBlockingReceive()

                if _message:
                    Logging.logChannelOp(_message.sender,
                                         _message.receiver,
                                         'receive',
                                         _message)

                    if isinstance(_message, Message.DownloadRequestMessage):
                        Logging.logProcessOp(processId=self.environment.procId,
                                             op="push",
                                             depQueue=self.downloadRequests,
                                             message=_message)
                        self.downloadRequests.put(_message)

                    if isinstance(_message, Message.AdvertisementMessage):
                        # request the sender for the segments
                        if not self.dataHandler.getSegment(_message.messageId):
                            _response = _message.getResponse()
                            Logging.logChannelOp(_message.sender,
                                                 _message.receiver,
                                                 'send',
                                                 _response)
                            self.environment.send(_response.receiver,
                                                  _response)

                    if isinstance(_message, Message.RequestMessage):
                        # add this to the request queue
                        Logging.logProcessOp(processId=self.environment.procId,
                                             op="push",
                                             depQueue=self.requestQueue,
                                             message=_message)
                        self.requestQueue.put(_message)

                    if isinstance(_message, Message.SegmentMessage):
                        Logging.logProcessOp(processId=self.environment.procId,
                                             op="receivedSegments")
                        self.dataHandler.addSegment(_message.messageId,
                                                    _message.content)
                        self.dataHandler.store(_message)

                    if isinstance(_message, Message.RequestResponseMessage):

                        assert(self.environment.processId ==
                               self.SegmentAssignProcId)

                        Logging.logProcessOP(processId=self.environment.procId,
                                             op="receivedDownloadConfirmation")

                        self.peers.removeBackLog(_message.sender)

                        if not _message.status:

                            self.segmentHandler.unassignSegment(
                                _message.messageId)

                            reqSegId = self.segmentHandler.getNextUnassigned()

                            requestSegment = RequestMessage(
                                self.SegmentAssignProcId,
                                reqSegId)

                            requestSegment.initMessage(
                                msgProperty=None,
                                msgContent=self.segmentHandler.getMetadata(
                                    reqSegId))

                            self.environment.send(feedback.fromId,
                                                  requestSegment)
                            self.segmentHandler.assignSegment(reqSegId)

            elif nonDetchoice == 1:
                self.handleRequestQueue()

            elif nonDetchoice == 2:
                self.handleDownloadRequestQueue()

            else:
                self.handleAdvertisementQueue()

    def microDownload(self):

        Logging.logProcessOp(processId=self.environment.procId,
                             op="startMicroNC")

        self.segmentHandler.downloadMetadata()
        self.peers.initPeers(self.environment.totalProc)

        while not self.segmentHandler.allAssigned():

            peerId = self.peers.leastBusyPeer()
            peerBackLog = self.peers.getBackLog(peerId)

            if peerBackLog < self.MAX_BACKLOG:

                requestSegmentId = self.segmentHandler.getNextUnassigned()
                requestSegment = Message.DownloadRequestMessage(
                    self.SegmentAssignProcId,
                    requestSegmentId,
                    peerId)
                requestSegment.initMessage(
                    msgProperty=None,
                    msgContent=self.segmentHandler.getMetadata(
                        requestSegmentId))

                self.environment.send(peerId, requestSegment)

                self.segmentHandler.assignSegment(requestSegmentId)
                self.peers.addBackLog(peerId)
                Logging.logChannelOp(chanFrom=self.SegmentAssignProcId,
                                     chanTo=peerId,
                                     op="send",
                                     message=requestSegment)

            else:
                Logging.info("Waiting for feedback")
\$\endgroup\$
  • 1
    \$\begingroup\$ Obligatory suggestion to follow the style guide (also, please review the indentation). \$\endgroup\$ – jonrsharpe Aug 2 '14 at 19:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.