Почему не происходит запись в MongoDB при нескольких потоках?

Question

[Ivan Yakushenko]

При изучении MongoDB наткнулся на возможность использования вместо сгенерированного значения _id упорядоченный (1, 2, 3...):

def get_next_sequence(collection, name):
    return collection.find_and_modify({'_id': name}, update={'$inc': {'seq': 1}}, new=True).get('seq')


def insert_in_db():
    client = MongoClient(mongo_url)
    db = client['']
    collection = db['']
    print(collection.insert_one({'_id': get_next_sequence(collection, 'userid'), 'value': f'{random.randint(10000, 2147483647)}'}))
    client.close()

Вариант работает, но меня смутила возможность его использования в несколько потоков. Если я правильно понимаю, то скрипт выполняет 2 запроса в БД:
1. Узнает значение последнего элемента
2. Присваивает новый элемент со значением +1

Если к нему обратятся одновременно несколько запросов? Начал проверять:

with Pool(processes=200) as pool:
    for _ in range(100000):
        pool.apply_async(insert_in_db)
    pool.close()
    pool.join()

Начал наращивать кол-во потоков по чуть-чуть - со значений в 50 и до 1000.
Вплоть до кол-во потоков в 200-250 проблем не было вообще никаких - все записывалось, после - начались пропуски записи. Особенно заметно после 400 потоков. При 1000 потоков из 100000 запросов на запись в базе оказалось только чуть больше 90000, при этом ошибок никаких нет. Сейчас у меня 2 предположения:
1. Не выдерживает сервер с MongoDB (3 ядра, 4гб ОЗУ). При работе скрипта были видны моменты, когда он просто останавливался на несколько секунд, иногда на 10-30 секунд. Видимо не мог подключиться. Соответственно некоторые потоки просто не могли подключиться и запись не происходила.
2. Обращаются несколько потоков, все получают информацию, что на данный момент, к примеру, 1389 записей в базе и пытаются записать запись под номером 1389. Смущает то, что ошибок записи нет в выводе, хотя по идеи должна быть.

Comments

[Taus]

Вы создаёте не потоки (которыми считаются Thread в Python), а процессы из multiprocessing. Каждый процесс порождает свой интерпретатор, что приведёт к безумной нагрузке.

[Ivan Yakushenko]

Taus, я знаю.

[Влад Григорьев]

Ivan Yakushenko, Если вы хотите писать в 1000 потоков и более в базу, то возможно вам стоит посмотреть в сторону redis

[Ivan Yakushenko]

Влад Григорьев, я не хочу писать в 1000 потоков и более. Я просто наткнулся на определенную ситуацию и хочу понять почему так происходит.

Answer 1

[Taus]

Вы не получаете результат выполнения в созданных процессах из pool.apply_async. Это плохая практика, потому что при выполнении кода в дочерних процессах могут быть исключения, которые стоит обрабатывать в основном процессе. Почитайте дополнительно документацию. Пример:

import multiprocessing

def f():
    raise ValueError()

with multiprocessing.Pool() as pool:
    for _ in range(10):
        pool.apply_async(f) # no errors

with multiprocessing.Pool() as pool:
    for _ in range(10):
        result = pool.apply_async(f)
        result.get(timeout=1) # raise ValueError

Поскольку у вас такой обработки, то можно предположить следующее. При некотором числе создаваемых процессов создание MongoClient или запросы .insert_one|.find_and_modify бросает исключение, связанное с превышением какого-то таймаута (посмотрите необязательные аргументы mongo_client и исключения)

Comments

[Ivan Yakushenko]

Ошибки следующего характера:

Traceback (most recent call last):
  File "/home/kshnkvn/.local/lib/python3.6/site-packages/pymongo/pool.py", line 973, in connect
    sock = _configured_socket(self.address, self.opts)
  File "/home/kshnkvn/.local/lib/python3.6/site-packages/pymongo/pool.py", line 848, in _configured_socket
    sock = _create_connection(address, options)
  File "/home/kshnkvn/.local/lib/python3.6/site-packages/pymongo/pool.py", line 829, in _create_connection
    raise err
  File "/home/kshnkvn/.local/lib/python3.6/site-packages/pymongo/pool.py", line 822, in _create_connection
    sock.connect(sa)
OSError: [Errno 99] Cannot assign requested address

Я не пойму, если у выставлю очень большие тайм-ауты - это может решить проблему?

[Taus]

Нет, не поможет. В какой-то момент у вас кончается число доступных сокетов из-за того, что при каждом вызове insert_in_db создаётся новый клиент (неправильно использование pymongo). Вероятно, что большинство из них висит в состоянии CLOSE_WAIT. Проверьте с помощь lsof | grep mongo.

[Ivan Yakushenko]

Taus, но у меня ведь есть client.close()

end all server sessions created by this client by sending one or more endSessions commands

Это же вроде должно предотвращать накопление соединений, разве нет?

[Taus]

Тоже считаю, что должно. Надо разбираться в деталях кто виноват. Очень вероятно, что TCP стек уже не успевает работать по причинам выше.

[Ivan Yakushenko]

Taus, основная проблема в том, что база хостится у cloudclusters, доступа по ssh нет, что сильно ограничивает возможности наблюдения за ней.

[Taus]

pymongo вызывает socket.close() внутри себя при вызове client.close(). Согласно документации socket.close() не гарантирует, что сокет будет немедленно закрыт, поэтому они и накапливаются.

Note
close() releases the resource associated with a connection but does not necessarily close the connection immediately. If you want to close the connection in a timely fashion, call shutdown() before close().

Поэтому хорошо подумайте и погуглите каким образом организовать работу pymongo+multiprocessing.