Pythonでネットワークプログラミング

import socket

"""
シングルプロセス・シングルスレッド・シングルクライアント
"""


def create_server_socket(port):
    # ソケットディスクリプタを生成
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # ソケットオプションでソケットの再利用フラグをONに設定
    # ONにしていないとクライアントと通信途中で中断した場合同じアドレスとポートでバインドできなくなります
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # ソケットにポート番号をバインド
    server_sock.bind(('', port))
    # アクセスバックログ(接続待ちのキュー数)を指定。
    server_sock.listen(5)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')

    while True:
        new_sock, (remote_host, remote_remport) = server_sock.accept()
        print('[FD:{}]Accept:{}:{}'.format(new_sock.fileno(), remote_host, remote_remport))
        data = new_sock.recv(512)
        print('[FD:{}]Recv:{}'.format(new_sock.fileno(), data))
        new_sock.send(data)
        new_sock.close()


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        server_sock.close()

import socket
import select

"""
シングルプロセス・シングルスレッド・マルチクライアント(select)
"""


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')
    # max_descriptor = 20
    descriptors = [server_sock]

    while True:
        print('Descriptor Count:{}'.format(len(descriptors) - 1))
        r, _, _ = select.select(descriptors, [], [])
        for sock in r:
            if sock == server_sock:
                # if len(descriptors) > max_descriptor:
                #     sock.close()
                new_sock, (remote_host, remote_remport) = sock.accept()
                print('[FD:{}]Accept:{}:{}'.format(new_sock.fileno(), remote_host, remote_remport))
                descriptors.append(new_sock)
            else:
                data = send_recv(sock)
                if not data:
                    sock.close()
                    descriptors.remove(sock)


def send_recv(sock):
    remote_host, remote_remport = sock.getpeername()
    print('[FD:{}]Client:{}:{}'.format(sock.fileno(), remote_host, remote_remport))

    data = sock.recv(512)
    if data == '':
        print('[FD:{}]Recv:EOF'.format(sock.fileno()))
        return None

    print('[FD:{}]Recv:{}'.format(sock.fileno(), data))
    sock.send(data)
    return data


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        server_sock.close()

import socket
import select

"""
シングルプロセス・シングルスレッド・マルチクライアント(epoll)
"""


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')

    epoll = select.epoll()
    epoll.register(server_sock.fileno(), select.EPOLLIN)

    # max_discriptor = 20
    discriptors = {}
    while True:
        events = epoll.poll()
        print('Discriptor Count: {}'.format(len(discriptors)))
        for fileno, event in events:
            if fileno == server_sock.fileno():
                new_sock, (remote_host, remote_remport) = server_sock.accept()
                # if len(discriptors) > max_discriptor:
                #     new_sock.close()
                epoll.register(new_sock.fileno(), select.EPOLLIN)
                discriptors[new_sock.fileno()] = new_sock
                print('[FD:{}]Accept: {}:{}'.format(new_sock.fileno(), remote_host, remote_remport))
            else:
                sock = discriptors[fileno]
                data = send_recv(sock)
                if not data:
                    sock.close()
                    del discriptors[fileno]


def send_recv(sock):
    remote_host, remote_remport = sock.getpeername()
    print('[FD:{}]Client:{}:{}'.format(sock.fileno(), remote_host, remote_remport))

    data = sock.recv(512)
    if data == b'':
        print('[FD:{}]Recv:EOF'.format(sock.fileno()))
        return None

    sock.send(data)
    return data


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        server_sock.close()

import os
import sys
import socket

"""
マルチプロセス・シングルスレッド・マルチクライアント
"""


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')
    while True:
        new_sock, (remote_host, remote_remport) = server_sock.accept()

        pid = os.fork()
        if pid == 0:
            # Child Precess
            server_sock.close()
            send_recv_loop(new_sock)
            new_sock.close()
            sys.exit()
        elif pid > 0:
            # Parent Precess
            print('[PID:{}]Accept:{}:{}'.format(os.getpid(), remote_host, remote_remport))
            new_sock.close()
        else:
            # Fork Fail
            new_sock.close()


def send_recv_loop(sock):
    while True:
        pid = os.getpid()
        data = sock.recv(512)
        remote_host, remote_remport = sock.getpeername()
        print('[PID:{}]Client:{}:{}'.format(pid, remote_host, remote_remport))

        if data == b'':
            print('[PID:{}]Recv:EOF'.format(pid))
            break

        print('[PID:{}]Recv:{}'.format(pid, data))
        sock.send(data)


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        server_sock.close()

import socket
import threading

"""
シングルプロセス・マルチスレッド・マルチクライアント
"""


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')
    while True:
        new_sock, (remote_host, remote_remport) = server_sock.accept()
        thread = threading.Thread(target=send_recv_thread, args=(new_sock,))
        thread.start()
        print('<{}>Accept:{}:{}'.format(thread.ident, remote_host, remote_remport))


def send_recv_thread(sock):
    while True:
        data = sock.recv(512)
        remote_host, remote_remport = sock.getpeername()
        ident = threading.currentThread().ident
        print('<{}>Client:{}:{}'.format(ident, remote_host, remote_remport))

        if data == b'':
            print('<{}>Recv:EOF'.format(ident))
            break

        print('<{}>Recv:{}'.format(ident, data))
        sock.send(data)
    sock.close()


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        server_sock.close()

import fcntl
import os
import sys
import socket

"""
マルチプロセス・シングルスレッド・マルチクライアント
"""


global_lock_fd = None


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    pid = os.getpid()
    while True:
        print('[PID:{}]Get Start'.format(pid))
        fcntl.lockf(global_lock_fd.fileno(), fcntl.LOCK_EX)
        print('[PID:{}]Get Lock'.format(pid))
        new_sock, (remote_host, remote_remport) = server_sock.accept()
        print('[PID:{}]Accept:{}:{}'.format(pid, remote_host, remote_remport))
        fcntl.lockf(global_lock_fd.fileno(), fcntl.LOCK_UN)
        print('[PID:{}]Unlock'.format(pid))
        send_recv_loop(new_sock)
        new_sock.close()


def send_recv_loop(sock):
    while True:
        pid = os.getpid()
        data = sock.recv(512)
        remote_host, remote_remport = sock.getpeername()
        print('[PID:{}]Client:{}:{}'.format(pid, remote_host, remote_remport))

        if data == b'':
            print('[PID:{}]Recv:EOF'.format(pid))
            break

        print('[PID:{}]Recv:{}'.format(pid, data))
        sock.send(data)


if __name__ == '__main__':
    server_sock = create_server_socket(7777)
    global_lock_fd = open('./mprocess.lock', 'w')
    os.unlink('./mprocess.lock')

    print('Ready For Accept')
    for i in range(10):
        pid = os.fork()
        if pid == 0:
            accept_loop(server_sock)
            sys.exit()

    try:
        os.wait()
    except KeyboardInterrupt:
        server_sock.close()
        global_lock_fd.close()

import time
import threading
import socket

"""
シングルプロセス・マルチスレッド・マルチクライアント
"""


LOCK = threading.Lock()


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_thread(server_sock):
    ident = threading.currentThread().ident
    while True:
        print('<{}>Start'.format(ident))
        LOCK.acquire()
        print('<{}>Get Lock'.format(ident))
        new_sock, (remote_host, remote_remport) = server_sock.accept()
        print('<{}>Accept:{}:{}'.format(ident, remote_host, remote_remport))
        LOCK.release()
        print('<{}>Release Lock'.format(ident))
        send_recv_loop(ident, new_sock)
        new_sock.close()


def send_recv_loop(ident, sock):
    while True:
        data = sock.recv(512)
        remote_host, remote_remport = sock.getpeername()
        print('<{}>Client:{}:{}'.format(ident, remote_host, remote_remport))

        if data == b'':
            print('<{}>Recv:EOF'.format(ident))
            break

        print('<{}>Recv:{}'.format(ident, data))
        sock.send(data)
    sock.close()


if __name__ == '__main__':
    server_sock = create_server_socket(7777)

    print('Ready For Accept')

    thread_list = []
    for _ in range(40):
        thread = threading.Thread(target=accept_thread, args=(server_sock,))
        thread.start()
        thread_list.append(thread)

    for thread in thread_list:
        thread.join()

    server_sock.close()

# -*- coding: utf-8 -*-

import socket
import threading

from Queue import Queue
import select

"""
シングルプロセス・マルチスレッド・マルチクライアント(select)
"""


class SendQueue(Queue):
    SENTINEL = object()

    def __iter__(self):
        while True:
            item = self.get()
            try:
                if item is self.SENTINEL:
                    yield None, None
                yield item
            finally:
                self.task_done()

    def close(self):
        self.put(self.SENTINEL)


send_queue = SendQueue()


def create_server_socket(port):
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('', port))
    server_sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return server_sock


def accept_loop(server_sock):
    print('Ready For Accept')
    max_descriptor = 20
    descriptors = [server_sock]

    while True:
        print('Descriptor Count:{}'.format(len(descriptors) - 1))
        r, _, _ = select.select(descriptors, [], [])
        for sock in r:
            if sock == server_sock:
                if len(descriptors) > max_descriptor:
                    sock.close()
                new_socket, (remote_host, remote_remport) = sock.accept()
                print('[FD:{}]Accept:{}:{}'.format(new_socket.fileno(), remote_host, remote_remport))
                descriptors.append(new_socket)
            else:
                remote_host, remote_remport = sock.getpeername()
                print('[FD:{}]Client:{}:{}'.format(sock.fileno(), remote_host, remote_remport))
                data = sock.recv(512)
                if data == b'':
                    print('[FD:{}]Recv:EOF'.format(sock.fileno()))
                    sock.close()
                    descriptors.remove(sock)
                else:
                    send_queue.put((sock, data))


def send_thread():
    current_thread = threading.current_thread()
    for sock, data in send_queue:
        if not sock:
            break
        print('[FD:{}]<{}>Recv:{}'.format(sock.fileno(), current_thread.ident, data))
        sock.send(data)


if __name__ == '__main__':
    server_sock = create_server_socket(7777)

    thread_count = 40
    thread_list = []
    for _ in range(thread_count):
        thread = threading.Thread(target=send_thread)
        thread.start()
        thread_list.append(thread)

    try:
        accept_loop(server_sock)
    except KeyboardInterrupt:
        for _ in range(thread_count):
            send_queue.close()
        for thread in thread_list:
            thread.join()

        send_queue.join()
        server_sock.close()

import asyncio
import socket

"""
シングルプロセス・シングルスレッド・マルチクライアント(asyncio)
"""


def create_server_socket(port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.setblocking(False)
    sock.bind(('', port))
    sock.listen(256)
    print('Server Run Port:{}'.format(port))
    return sock


async def accept(loop, sock):
    print('Ready For Accept')

    while True:
        new_socket, (remote_host, remote_remport) = await loop.sock_accept(sock)
        new_socket.setblocking(False)
        print('[FD:{}]Accept:{}:{}'.format(new_socket.fileno(), remote_host, remote_remport))
        asyncio.ensure_future(recv_send(loop, new_socket))


async def recv_send(loop, sock):
    remote_host, remote_remport = sock.getpeername()
    print('[FD:{}]Client:{}:{}'.format(sock.fileno(), remote_host, remote_remport))

    while True:
        data = await loop.sock_recv(sock, 512)
        if data == b'':
            print('[FD:{}]Recv:EOF'.format(sock.fileno()))
            sock.close()
            break

        print('[FD:{}]Recv:{}'.format(sock.fileno(), data))
        await loop.sock_sendall(sock, data)


if __name__ == '__main__':
    event_loop = asyncio.SelectorEventLoop()
    asyncio.set_event_loop(event_loop)
    server_sock = create_server_socket(7777)

    try:
        event_loop.run_until_complete(accept(event_loop, server_sock))
    except KeyboardInterrupt:
        event_loop.close()
        server_sock.close()

# -*- coding: utf-8 -*-

import socket
import threading
import time

WORKER_COUNT = 150
ACTION_COUNT = 100
SEND_RECV_COUNT = 50


class WorkerInfo(object):

    def __init__(self):
        self.thread_name = ''
        self.is_active = True
        self.connect_speed = 0.0
        self.send_recv_speed = 0.0
        self.action_count = 0

    def create_send_msg(self, send_recv_count):
        return b'{}:{}:{}'.format(self.thread_name, self.action_count, send_recv_count)

worker_info_list = []


def create_client_socket(port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(('', port))
    return sock


def worker_send_recv(worker_info):
    # connect
    connect_start_time = time.time()
    sock = create_client_socket(7777)
    worker_info.connect_speed += time.time() - connect_start_time

    # send and recv
    send_recv_start_time = time.time()
    for cnt in range(SEND_RECV_COUNT, 1):
        send_msg = worker_info.create_send_msg(cnt)
        sock.send(send_msg)
        recv_msg = sock.recv(512)
        if recv_msg != send_msg:
            print('error')
    sock.close()
    worker_info.send_recv_speed += time.time() - send_recv_start_time


def worker(worker_idx):
    worker_info = worker_info_list[worker_idx]
    worker_info.thread_name = threading.current_thread().name
    print(worker_info.thread_name)

    for cnt in range(ACTION_COUNT):
        worker_info.action_count = cnt
        worker_send_recv(worker_info)

    # Average calculation
    worker_info.is_active = False
    worker_info.connect_speed /= ACTION_COUNT
    worker_info.send_recv_speed /= ACTION_COUNT


if __name__ == '__main__':
    # Test Start
    thread_list = []
    for idx in range(WORKER_COUNT):
        worker_info_list.append(WorkerInfo())
        thread = threading.Thread(target=worker, name='thread:{}'.format(idx), args=(idx,))
        thread.start()
        thread_list.append(thread)

    for thread in thread_list:
        thread.join()

    # All Thread Average calculation
    connect_speed = 0
    send_recv_speed = 0
    for idx in range(WORKER_COUNT):
        _worker_info = worker_info_list[idx]
        connect_speed += _worker_info.connect_speed
        send_recv_speed += _worker_info.send_recv_speed

    # Result
    connect_speed = round((connect_speed / WORKER_COUNT), 5)
    send_recv_speed = round((send_recv_speed / WORKER_COUNT), 5)
    total_speed = connect_speed + send_recv_speed
    print('connect_speed: {}'.format(connect_speed))
    print('send_recv_speed: {}'.format(send_recv_speed))
    print('total_speed: {}'.format(total_speed))