Руководство по программированию сокетов на Python. Клиент, сервер и несколько соединений

# echo-server.py

import socket

HOST = "127.0.0.1"  # Standard loopback interface address (localhost)
PORT = 65432  # Port to listen on (non-privileged ports are > 1023)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    conn, addr = s.accept()
    with conn:
        print(f"Connected by {addr}")
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    pass  # Use the socket object without calling s.close().

# echo-server.py

# ...

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    # ...

# echo-server.py

# ...

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    conn, addr = s.accept()
    # ...

# echo-server.py

# ...

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    conn, addr = s.accept()
    with conn:
        print(f"Connected by {addr}")
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)

# echo-client.py

import socket

HOST = "127.0.0.1"  # The server's hostname or IP address
PORT = 65432  # The port used by the server

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, PORT))
    s.sendall(b"Hello, world")
    data = s.recv(1024)

print(f"Received {data!r}")

$ python echo-server.py

# echo-server.py

# ...

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    conn, addr = s.accept()
    with conn:
        print(f"Connected by {addr}")
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)

$ python echo-client.py 
Received b'Hello, world'

$ python echo-server.py 
Connected by ('127.0.0.1', 64623)

$ netstat -an
Active Internet connections (including servers)
Proto Recv-Q Send-Q  Local Address          Foreign Address        (state)
tcp4       0      0  127.0.0.1.65432        *.*                    LISTEN

$ netstat -an
Active Internet connections (including servers)
Proto Recv-Q Send-Q  Local Address          Foreign Address        (state)
tcp4       0      0  *.65432                *.*                    LISTEN

$ lsof -i -n
COMMAND     PID   USER   FD   TYPE   DEVICE SIZE/OFF NODE NAME
Python    67982 nathan    3u  IPv4 0xecf272      0t0  TCP *:65432 (LISTEN)

$ python echo-client.py 
Traceback (most recent call last):
  File "./echo-client.py", line 9, in <module>
    s.connect((HOST, PORT))
ConnectionRefusedError: [Errno 61] Connection refused

Разбор взаимодействия клиента и сервера

Посмотрите подробности взаимодействия клиента и сервера:

При работе с интерфейсом «внутренней петли» (IPv4 127.0.0.1 или IPv6 ::1) данные никогда не покидают хост и не касаются внешней сети. Выше интерфейс внутренней петли находится внутри хоста. Такова его природа: подключения и данные, которые пропускаются через него, локальны для хоста. Поэтому интерфейс внутренней петли и IP-адреса 127.0.0.1 и ::1 и называются localhost.

В приложениях этот интерфейс используется для взаимодействия с другими процессами, которые запускаются на хосте, а также в целях безопасности и предоставления изолированности от внешней сети. Внутренний, доступный только изнутри хоста, этот интерфейс недоступен извне.

Посмотреть на это в действии можно, если у вас есть сервер приложений со своей, закрытой базой данных. Если эта БД не используется на других серверах, наверное, она конфигурируется на прослушивание подключений только в интерфейсе localhost. Если так, то другим хостам сети нельзя подключиться к нему.

Если не использовать в приложениях адреса 127.0.0.1 или ::1, вероятно, отличный от них адрес привяжется к интерфейсу Ethernet, подключённому к внешней сети. Это шлюз к другим хостам за пределами localhost:

Поосторожнее там! Это неприглядный, жестокий мир. Прежде чем отважиться покинуть безопасный localhost, обязательно прочитайте раздел «Использование имён хостов». Там есть примечания по безопасности, которое применяется, даже если вы используете не имена хостов, а только IP-адреса.

Обработка нескольких подключений

У эхо-сервера определённо имеются ограничения. Самое большое — им обслуживается только один клиент, после чего работа завершается. У эхо-клиента такое же ограничение, но имеется дополнительная проблема. Когда в клиенте используется s.recv(), из b'Hello, world' можно вернуть только один байт (b'H'):

# echo-client.py

# ...

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, PORT))
    s.sendall(b"Hello, world")
    data = s.recv(1024)

print(f"Received {data!r}")

Аргумент bufsize (1024) — это максимальный объём данных, получаемых одновременно. Он не означает, что из .recv() вернётся 1024 байт.

Поведение метода .send() аналогичное: в нём возвращается число отправленных байтов, которое может оказаться меньше размера переданных данных. Это нужно проверить и, чтобы отправить данные, вызвать .send() столько раз, сколько потребуется:

«В приложениях нужно проверять, что отправлены все данные. Если передана только их часть, нужно пробовать доставить оставшиеся данные». (Источник)

В примере выше этого удалось избежать благодаря .sendall():

«В отличие от send(), в этом методе отправка байт данных продолжается, пока не будут отправлены все данные или не возникнет ошибка. В случае успеха метод возвращает None». (Источник)

Здесь возникают две проблемы:

• Обработка нескольких подключений одновременно.
• Необходимость вызывать .send() и .recv(), пока все данные не будут отправлены или получены.

Что можно сделать? В плане конкурентности подходов много. Один из популярных — применение асинхронного ввода-вывода. asyncio появился в стандартной библиотеке в Python 3.4. Традиционное решение — использовать потоки.

Проблема конкурентности заключается в сложности её правильного понимания. Здесь нужно учитывать и остерегаться многих нюансов. Стóит проявиться одному из них, и приложение внезапно может просто-напросто выйти из строя.

Это не должно отбить желание осваивать и применять конкурентное программирование. Если приложение нужно масштабировать, использование более одного процессора или ядра необходимо. Но это руководство опирается на дедушку системных вызовов – более традиционый метод .select(), объяснить который проще.

Метод .select() проверяет завершение ввода-вывода более чем на одном сокете. Вызовите его, чтобы посмотреть, у каких сокетов ввод-вывод готов для чтения и/или для записи. Кроме .select(), есть ещё кое-что: чтобы использовать самую эффективную реализацию независимо от ОС, будет применяться стандартный модуль selectors:

«Этим модулем обеспечивается высокоуровневое и эффективное мультиплексирование ввода-вывода, основанное на примитивах модуля select. Пользователям рекомендуется применять этот модуль, если нет необходимости точно управлять примитивами, которые используются на уровне ОС». (Источник)

Тем не менее, применяя .select(), вы не сможете обеспечить конкурентное выполнение. В то же время, в зависимости от рабочей нагрузки, этот подход всё равно может оказаться достаточно быстрым. Это определяется количеством поддерживаемых в приложении клиентов и тем, что именно в нём должно делаться при обслуживании запроса.

Модуль asyncio для управления задачами использует однопоточную кооперативную многозадачность и цикл событий. С помощью .select() вы напишете собственную версию цикла событий, пусть даже проще и синхроннее. При использовании нескольких потоков, даже если присутствует конкурентность, сейчас приходится применять GIL (глобальную блокировку интерпретатора) с CPython и PyPy. Этим, так или иначе, ограничивается объём работы, выполняемой параллельно.

Всё это свидетельствует о том, что использование .select() может быть идеальным выбором. Не думайте, что придётся задействовать asyncio, потоки или новейшую асинхронную библиотеку. В сетевом приложении задача обычно, так или иначе, ограничена вводом-выводом: это может быть ожидание в локальной сети конечных точек на другой стороне сети, ожидание записи на диск и т. д.

Если вы получаете запросы от клиентов, которыми инициируется связанная с ЦП работа, обратите внимание на модуль concurrent.futures. В нём есть класс ProcessPoolExecutor. Этот класс для асинхронного выполнения вызовов использует пул процессов.

Если вы задействуете несколько процессов, в операционной системе может планироваться параллельное выполнение кода Python на нескольких процессорах или ядрах без глобальной блокировки интерпретатора. Идеи и вдохновение можно почерпнуть из доклада на PyCon John Reese — Thinking Outside the GIL with AsyncIO and Multiprocessing — PyCon 2018 («Джон Риз. AsyncIO и многопроцессорная обработка: выходя за рамки глобальной блокировки интерпретатора. PyCon 2018»).

В следующем разделе вы увидите примеры сервера и клиента с решением этих проблем. Чтобы одновременно обрабатывать несколько подключений и вызывать .send() и .recv() нужное количество раз, в примерах используется .select().

Клиент и сервер с несколькими подключениями

В следующих разделах вы напишете сервер и клиента, где несколько подключений обрабатываются с помощью объекта selector из модуля selectors.

Сервер с несколькими подключениями

Обратимся сначала к серверу с несколькими подключениями. В первой части настраивается прослушиваемый сокет:

# multiconn-server.py

import sys
import socket
import selectors
import types

sel = selectors.DefaultSelector()

# ...

host, port = sys.argv[1], int(sys.argv[2])
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
lsock.bind((host, port))
lsock.listen()
print(f"Listening on {(host, port)}")
lsock.setblocking(False)
sel.register(lsock, selectors.EVENT_READ, data=None)

Самое большое отличие этого сервера от эхо-сервера — вызов настройки сокета в неблокируемом режиме внутри lsock.setblocking(False). Вызовы в этот сокет больше не будут блокирующими. Когда этот вызов применяется вместе с sel.select(), можно ожидать события в одном или нескольких сокетах, а затем считывать и записывать данные по мере их готовности. Демонстрацию вы увидите ниже.

С помощью sel.register() сокет регистрируется, а через sel.select() в нём отслеживаются нужные события. Для прослушиваемого сокета, например, нужны события чтения, то есть selectors.EVENT_READ.

Чтобы сохранять любые произвольные данные, которые бы понадобились вместе с сокетом, и отслеживать, чтó в нём отправлено и получено, используется data, возвращаемый вместе с .select().

Вот цикл событий:

# multiconn-server.py

# ...

try:
    while True:
        events = sel.select(timeout=None)
        for key, mask in events:
            if key.data is None:
                accept_wrapper(key.fileobj)
            else:
                service_connection(key, mask)
except KeyboardInterrupt:
    print("Caught keyboard interrupt, exiting")
finally:
    sel.close()

sel.select(timeout=None) realpython.com/python-sockets//#blocking-calls, пока не будет сокетов, готовых для ввода-вывода. В нём возвращается список кортежей — по одному для каждого сокета. В каждом кортеже есть key и mask. key — это namedtuple [именованный кортеж] SelectorKey, в котором содержится атрибут fileobj. key.fileobj — это объект сокета, а mask— маска события готовых операций.

Если key.data будет None, то понятно, что это из прослушиваемого сокета, и подключение нужно принять. Чтобы получить новый объект сокета и зарегистрировать его с селектором, вызывается ваша функция accept_wrapper(). Она показана чуть ниже.

Если key.data не None, то понятно, что это сокет уже принятого клиента, и его нужно обслуживать. Затем вызывается service_connection() с аргументами key и mask. Это всё, что нужно для работы с сокетом.

Вот что делается в функции accept_wrapper():

# multiconn-server.py

# ...

def accept_wrapper(sock):
    conn, addr = sock.accept()  # Should be ready to read
    print(f"Accepted connection from {addr}")
    conn.setblocking(False)
    data = types.SimpleNamespace(addr=addr, inb=b"", outb=b"")
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    sel.register(conn, events, data=data)

# ...

Прослушиваемый сокет зарегистрирован для события selectors.EVENT_READ, поэтому он должен быть готов к считыванию. Вызывается sock.accept(), а затем, для перевода сокета в неблокируемый режим, – conn.setblocking(False).

И это – основная цель данной версии сервера, ведь нужно, чтобы он не блокировался. Если сервер заблокируется, то он весь остановится до возврата. Это означает, что другие сокеты остаются в ожидании, даже если сервер не используется активно – вот и жутковатое «зависания», в котором серверу находиться нежелательно.

Дальше при помощи SimpleNamespace создаётся объект для хранения данных, которые должны идти вместе с сокетом. Поскольку вам нужно знать, когда подключение клиента готово для чтения и записи, оба этих события задаются с помощью побитового «ИЛИ»:

# multiconn-server.py

# ...

def accept_wrapper(sock):
    conn, addr = sock.accept()  # Should be ready to read
    print(f"Accepted connection from {addr}")
    conn.setblocking(False)
    data = types.SimpleNamespace(addr=addr, inb=b"", outb=b"")
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    sel.register(conn, events, data=data)

# ...

Маска events, сокет и объекты данных затем передаются в sel.register().

Теперь посмотрите, как обрабатывается готовое подключение клиента в service_connection():

# multiconn-server.py

# ...

def service_connection(key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            data.outb += recv_data
        else:
            print(f"Closing connection to {data.addr}")
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print(f"Echoing {data.outb!r} to {data.addr}")
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]

# ...

Вот сердце простого сервера с несколькими подключениями. key — это namedtuple, возвращаемый из .select(), в котором содержится объект сокета fileobj и объект данных. В mask находятся готовые события.

Если сокет готов к считыванию, mask & selectors.EVENT_READ окажется равным True, поэтому вызывается sock.recv(). Любые считанные данные добавляются в data.outb, их можно отправить позже.

Обратите внимание на блок else:. В нём проверяется, не получены ли данные:

# multiconn-server.py

# ...

def service_connection(key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            data.outb += recv_data
        else:
            print(f"Closing connection to {data.addr}")
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print(f"Echoing {data.outb!r} to {data.addr}")
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]

# ...

Если данные не получены, значит, сокет на клиенте закрыли, поэтому на сервере нужно сделать то же самое. Но не забудьте перед закрытием вызвать sel.unregister(), чтобы сокет не отслеживался через .select().

Когда сокет готов к записи (работоспособный сокет всегда должен быть готов к ней), любые полученные данные, которые хранятся в data.outb, передаются клиенту с помощью sock.send(). Затем отправленные байты удаляются из буфера для пересылки:

# multiconn-server.py

# ...

def service_connection(key, mask):

    # ...

    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print(f"Echoing {data.outb!r} to {data.addr}")
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]

# ...

В методе .send() возвращается число отправленных байтов, с помощью этого числа позже байты можно отбросить при помощи нотации среза в буфере .outb.

Клиент с несколькими подключениям

Теперь обратимся к multiconn-client.py, клиенту с несколькими подключениями. Он очень похож на сервер. Только вместо прослушивания подключений он начинается с их инициирования через start_connections():

# multiconn-client.py

import sys
import socket
import selectors
import types

sel = selectors.DefaultSelector()
messages = [b"Message 1 from client.", b"Message 2 from client."]

def start_connections(host, port, num_conns):
    server_addr = (host, port)
    for i in range(0, num_conns):
        connid = i + 1
        print(f"Starting connection {connid} to {server_addr}")
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(False)
        sock.connect_ex(server_addr)
        events = selectors.EVENT_READ | selectors.EVENT_WRITE
        data = types.SimpleNamespace(
            connid=connid,
            msg_total=sum(len(m) for m in messages),
            recv_total=0,
            messages=messages.copy(),
            outb=b"",
        )
        sel.register(sock, events, data=data)

# ...

Значение num_conns считывается из командной строки. Это количество создаваемых подключений к серверу. Аналогично серверу каждый сокет настраивается на неблокируемый режим.

С помощью .connect() сразу выдаётся исключение BlockingIOError, поэтому вместо него используется метод .connect_ex(), где изначально возвращается индикатор ошибки errno.EINPROGRESS, а не выдаётся исключение, из-за которого подключение затрудняется. Когда подключение завершено, готовый к чтению и записи сокет возвращается из .select().

После настройки сокета для создания сохраняемых с ним данных используется SimpleNamespace. Сообщения, отправляемые от клиента на сервер, копируются с помощью messages.copy(), потому что при каждом подключении вызывается socket.send() и список меняется. В объекте data хранится всё необходимое для отслеживания того, что именно на клиенте нужно отправить, что именно отправлено и получено (в том числе общее количество байтов в сообщениях).

Вот изменения [добавленные и удалённые строки, + и -] из серверного service_connection() для версии на клиенте:

 def service_connection(key, mask):
     sock = key.fileobj
     data = key.data
     if mask & selectors.EVENT_READ:
         recv_data = sock.recv(1024)  # Should be ready to read
         if recv_data:
-            data.outb += recv_data
+            print(f"Received {recv_data!r} from connection {data.connid}")
+            data.recv_total += len(recv_data)
-        else:
-            print(f"Closing connection {data.connid}")
+        if not recv_data or data.recv_total == data.msg_total:
+            print(f"Closing connection {data.connid}")
             sel.unregister(sock)
             sock.close()
     if mask & selectors.EVENT_WRITE:
+        if not data.outb and data.messages:
+            data.outb = data.messages.pop(0)
         if data.outb:
-            print(f"Echoing {data.outb!r} to {data.addr}")
+            print(f"Sending {data.outb!r} to connection {data.connid}")
             sent = sock.send(data.outb)  # Should be ready to write
             data.outb = data.outb[sent:]

По сути, то же самое, но с важным отличием. В клиенте отслеживается количество полученных от сервера байтов, так что эту сторону подключения можно закрыть. Когда это обнаруживается на сервере, другая сторона подключения тоже закрывается.

Cервер при этом зависит от поведения клиента, когда ожидается, что та сторона подключения закроется после отправки сообщений. Если клиент не закроется, подключение на сервере останется открытым. На практике может понадобиться защита от этого путём реализации на сервере тайм-аута для предотвращения накопления клиентских подключений, если в них по истечении определённого времени не отправляется запрос.

Запуск клиента и сервера с несколькими подключениями

Пора запускать multiconn-server.py и multiconn-client.py. И там и там используются аргументы командной строки. Можно запустить без аргументов и посмотреть варианты.

Для сервера передайте номера host и port:

$ python multiconn-server.py
Usage: multiconn-server.py <host> <port>

А для клиента передайте num_connections, то есть количество создаваемых подключений к серверу:

$ python multiconn-client.py
Usage: multiconn-client.py <host> <port> <num_connections>

Вот вывод сервера при прослушивании в интерфейсе «внутренней петли» в порте 65432:

$ python multiconn-server.py 127.0.0.1 65432
Listening on ('127.0.0.1', 65432)
Accepted connection from ('127.0.0.1', 61354)
Accepted connection from ('127.0.0.1', 61355)
Echoing b'Message 1 from client.Message 2 from client.' to ('127.0.0.1', 61354)
Echoing b'Message 1 from client.Message 2 from client.' to ('127.0.0.1', 61355)
Closing connection to ('127.0.0.1', 61354)
Closing connection to ('127.0.0.1', 61355)

А вот вывод клиента при создании в нём двух подключений к серверу:

$ python multiconn-client.py 127.0.0.1 65432 2
Starting connection 1 to ('127.0.0.1', 65432)
Starting connection 2 to ('127.0.0.1', 65432)
Sending b'Message 1 from client.' to connection 1
Sending b'Message 2 from client.' to connection 1
Sending b'Message 1 from client.' to connection 2
Sending b'Message 2 from client.' to connection 2
Received b'Message 1 from client.Message 2 from client.' from connection 1
Closing connection 1
Received b'Message 1 from client.Message 2 from client.' from connection 2
Closing connection 2

Отлично! Клиент и сервер с несколькими подключениями запущен. В следующем разделе вы расширите применение этого примера.

• Профессия Fullstack-разработчик на Python (15 месяцев)
• Профессия Data Scientist (24 месяца)