<commit_before>#!/usr/bin/env python 
#
# Cloudlet Infrastructure for Mobile Computing
#
#   Author: <NAME> <<EMAIL>>
#
#   Copyright (C) 2011-2013 Carnegie Mellon University
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
#

import sys
sys.path.insert(0, "../")
import time
import SocketServer
import socket
import select
import traceback
import struct
import json
import Queue
import threading
import pprint
from optparse import OptionParser

from control.config import ServiceMeta as SERVICE_META
from control.upnp_client import UPnPClient
import logging


class tempLOG(object):
    def info(self, data):
        sys.stdout.write("INFO\t" + data + "\n")
    def debug(self, data):
        sys.stdout.write("DEBUG\t" + data + "\n")
    def error(self, data):
        sys.stderr.write("ERROR\t" + data + "\n")


LOG = tempLOG()
upnp_client = UPnPClient()
output_queue = Queue.Queue()


class UCommConst(object):
    RESULT_RECEIVE_PORT     =   10120


class UCommError(Exception):
    pass


def process_command_line(argv):
    VERSION = 'gabriel discovery'
    DESCRIPTION = "Gabriel service discovery"

    parser = OptionParser(usage='%prog [option]', version=VERSION,
            description=DESCRIPTION)

    parser.add_option(
            '-s', '--address', action='store', dest='address',
            help="(IP address:port number) of directory server")
    settings, args = parser.parse_args(argv)
    if len(args) >= 1:
        parser.error("invalid arguement")

    if hasattr(settings, 'address') and settings.address is not None:
        if settings.address.find(":") == -1:
            parser.error("Need address and port. Ex) 10.0.0.1:8081")
    return settings, args


def get_service_list(argv):
    settings, args = process_command_line(sys.argv[1:])
    service_list = None
    if settings.address is None:
        upnp_client.start()
        upnp_client.join()
        service_list = upnp_client.service_list
    else:
        import urllib2
        ip_addr, port = settings.address.split(":", 1)
        port = int(port)
        meta_stream = urllib2.urlopen("http://%s:%d/" % (ip_addr, port))
        meta_raw = meta_stream.read()
        service_list = json.loads(meta_raw)
    pstr = pprint.pformat(service_list)
    LOG.info("Gabriel Server :")
    LOG.info(pstr)
    return service_list


class ResultForwardingClient(threading.Thread):
    """
    This client will forward all the result received from the client
    """

    def __init__(self, control_address, output_queue):
        self.stop = threading.Event()
        self.output_queue = output_queue
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.sock.connect(control_address)
        threading.Thread.__init__(self, target=self.forward)

    def forward(self):
        output_list = [self.sock.fileno()]
        error_list = [self.sock.fileno()]

        LOG.info("Start forwardning data")
        try:
            while(not self.stop.wait(0.00001)):
                inputready, outputready, exceptready = \
                        select.select([], output_list, error_list, 0)
                for s in outputready:
                    self._handle_result_output(s)
                for s in exceptready:
                    break
        except Exception as e:
            LOG.warning(traceback.format_exc())
            LOG.warning("%s" % str(e))
            LOG.debug("Result Forwading thread terminated")
        self.terminate()

    def terminate(self):
        self.stop.set()
        if self.sock is not None:
            self.sock.close()

    def _handle_result_output(self, socket_fd):
        while self.output_queue.empty() is False:
            try:
                return_data = self.output_queue.get_nowait()
                packet = struct.pack("!I%ds" % len(return_data),
                        len(return_data), return_data)
                self.sock.sendall(packet)
                LOG.info("forward the result: %s" % return_data)
            except Queue.Empty as e:
                pass


class UCommServerHandler(SocketServer.StreamRequestHandler, object):
    def setup(self):
        super(UCommServerHandler, self).setup()

    def _recv_all(self, recv_size):
        data = ''
        while len(data) < recv_size:
            tmp_data = self.request.recv(recv_size - len(data))
            if tmp_data == None or len(tmp_data) == 0:
                raise UCommError("Socket is closed")
            data += tmp_data
        return data

    def handle(self):
        try:
            LOG.info("new Offlaoding Engine is connected")
            socket_fd = self.request.fileno()
            input_list = [socket_fd]
            while True:
                inputready, outputready, exceptready = \
                        select.select(input_list, [], [], 0)
                for insocket in inputready:
                    if insocket == socket_fd:
                        self._handle_input_stream()
                time.sleep(0.001)
        except Exception as e:
            #LOG.debug(traceback.format_exc())
            LOG.debug("%s" % str(e))
            LOG.info("Offloading engine is disconnected")
            self.terminate()

    def _handle_input_stream(self):
        header_size = struct.unpack("!I", self._recv_all(4))[0]
        header_data = self._recv_all(header_size)
        print "received : %s" % header_data


class UCommServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
    stopped = False

    def __init__(self, port, handler):
        server_address = ('0.0.0.0', port)
        self.allow_reuse_address = True
        self.handler = handler
        try:
            SocketServer.TCPServer.__init__(self, server_address, handler)
        except socket.error as e:
            sys.stderr.write(str(e))
            sys.stderr.write("Check IP/Port : %s\n" % (str(server_address)))
            sys.exit(1)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        LOG.info("* UCOMM server configuration")
        LOG.info(" - Open TCP Server at %s" % (str(server_address)))
        LOG.info(" - Disable nagle (No TCP delay)  : %s" %
                str(self.socket.getsockopt(
                    socket.IPPROTO_TCP,
                    socket.TCP_NODELAY)))
        LOG.info("-" * 50)

    def serve_forever(self):
        while not self.stopped:
            self.handle_request()
<commit_msg>ucomm-control VM is working<commit_after>#!/usr/bin/env python 
#
# Cloudlet Infrastructure for Mobile Computing
#
#   Author: <NAME> <<EMAIL>>
#
#   Copyright (C) 2011-2013 Carnegie Mellon University
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
#

import sys
sys.path.insert(0, "../")
import time
import SocketServer
import socket
import select
import traceback
import struct
import json
import Queue
import threading
import pprint
from optparse import OptionParser

from control.config import ServiceMeta as SERVICE_META
from control.upnp_client import UPnPClient
import logging


class tempLOG(object):
    def info(self, data):
        sys.stdout.write("INFO\t" + data + "\n")
    def warning(self, data):
        sys.stdout.write("warning\t" + data + "\n")
    def debug(self, data):
        sys.stdout.write("DEBUG\t" + data + "\n")
    def error(self, data):
        sys.stderr.write("ERROR\t" + data + "\n")


LOG = tempLOG()
upnp_client = UPnPClient()
output_queue = Queue.Queue()


class UCommConst(object):
    RESULT_RECEIVE_PORT     =   10120


class UCommError(Exception):
    pass


def process_command_line(argv):
    VERSION = 'gabriel discovery'
    DESCRIPTION = "Gabriel service discovery"

    parser = OptionParser(usage='%prog [option]', version=VERSION,
            description=DESCRIPTION)

    parser.add_option(
            '-s', '--address', action='store', dest='address',
            help="(IP address:port number) of directory server")
    settings, args = parser.parse_args(argv)
    if len(args) >= 1:
        parser.error("invalid arguement")

    if hasattr(settings, 'address') and settings.address is not None:
        if settings.address.find(":") == -1:
            parser.error("Need address and port. Ex) 10.0.0.1:8081")
    return settings, args


def get_service_list(argv):
    settings, args = process_command_line(sys.argv[1:])
    service_list = None
    if settings.address is None:
        upnp_client.start()
        upnp_client.join()
        service_list = upnp_client.service_list
    else:
        import urllib2
        ip_addr, port = settings.address.split(":", 1)
        port = int(port)
        meta_stream = urllib2.urlopen("http://%s:%d/" % (ip_addr, port))
        meta_raw = meta_stream.read()
        service_list = json.loads(meta_raw)
    pstr = pprint.pformat(service_list)
    LOG.info("Gabriel Server :")
    LOG.info(pstr)
    return service_list


class ResultForwardingClient(threading.Thread):
    """
    This client will forward all the result received from the client
    """

    def __init__(self, control_address, output_queue):
        self.stop = threading.Event()
        self.output_queue = output_queue
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.sock.connect(control_address)
        threading.Thread.__init__(self, target=self.forward)

    def forward(self):
        output_list = [self.sock.fileno()]
        error_list = [self.sock.fileno()]

        LOG.info("Start forwardning data")
        try:
            while(not self.stop.wait(0.00001)):
                inputready, outputready, exceptready = \
                        select.select([], output_list, error_list, 0)
                for s in outputready:
                    self._handle_result_output()
                for s in exceptready:
                    break
        except Exception as e:
            LOG.warning(traceback.format_exc())
            LOG.warning("%s" % str(e))
            LOG.debug("Result Forwading thread terminated")
        self.terminate()

    def terminate(self):
        self.stop.set()
        if self.sock is not None:
            self.sock.close()

    def _handle_result_output(self):
        while self.output_queue.empty() is False:
            try:
                return_data = self.output_queue.get_nowait()
                packet = struct.pack("!I%ds" % len(return_data),
                        len(return_data), return_data)
                self.sock.sendall(packet)
                LOG.info("forward the result: %s" % return_data)
            except Queue.Empty as e:
                pass


class UCommServerHandler(SocketServer.StreamRequestHandler, object):
    def setup(self):
        super(UCommServerHandler, self).setup()

    def _recv_all(self, recv_size):
        data = ''
        while len(data) < recv_size:
            tmp_data = self.request.recv(recv_size - len(data))
            if tmp_data == None or len(tmp_data) == 0:
                raise UCommError("Socket is closed")
            data += tmp_data
        return data

    def handle(self):
        try:
            LOG.info("new Offlaoding Engine is connected")
            socket_fd = self.request.fileno()
            input_list = [socket_fd]
            while True:
                inputready, outputready, exceptready = \
                        select.select(input_list, [], [], 0)
                for insocket in inputready:
                    if insocket == socket_fd:
                        self._handle_input_stream()
                time.sleep(0.001)
        except Exception as e:
            #LOG.debug(traceback.format_exc())
            LOG.debug("%s" % str(e))
            LOG.info("Offloading engine is disconnected")
            self.terminate()

    def _handle_input_stream(self):
        global output_queue
        header_size = struct.unpack("!I", self._recv_all(4))[0]
        header_data = self._recv_all(header_size)
        output_queue.put(header_data)


class UCommServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
    stopped = False

    def __init__(self, port, handler):
        server_address = ('0.0.0.0', port)
        self.allow_reuse_address = True
        self.handler = handler
        try:
            SocketServer.TCPServer.__init__(self, server_address, handler)
        except socket.error as e:
            sys.stderr.write(str(e))
            sys.stderr.write("Check IP/Port : %s\n" % (str(server_address)))
            sys.exit(1)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        LOG.info("* UCOMM server configuration")
        LOG.info(" - Open TCP Server at %s" % (str(server_address)))
        LOG.info(" - Disable nagle (No TCP delay)  : %s" %
                str(self.socket.getsockopt(
                    socket.IPPROTO_TCP,
                    socket.TCP_NODELAY)))
        LOG.info("-" * 50)

    def serve_forever(self):
        while not self.stopped:
            self.handle_request()

{'fraction_non_alphanumeric': 0.10667780795209247, 'fraction_numerical': 0.008216698001531927, 'mean_word_length': 2.3509099393373774, 'pattern_counts': {'":': 4, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3594799', 'n_tokens_mistral': 4328, 'n_tokens_neox': 4148, 'n_words': 1035}