mrq/tests/test_memoryleaks.py at 8cd35a92be9e01d20c43bc2f6cc7399602f2f6ea · pricingassistant/mrq · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
from __future__ import print_function
from builtins import range
import time
import pytest
import os


def test_max_memory_restart(worker):

    N = 10

    worker.start(
        flags=" --processes 1 --greenlets 1 --max_memory 50 --report_interval 1")

    worker.send_tasks(
        "tests.tasks.general.Leak",
        [{"size": 1000000, "sleep": 1} for _ in range(N)],
        queue="default",
        block=False
    )

    i = 0
    while worker.mongodb_jobs.mrq_jobs.find({"status": "success"}).count() != N:
        time.sleep(1)
        i += 1
        if i % 5 == 0:
            os.system("ps -ef")

    # We must have been restarted at least once.
    assert worker.mongodb_jobs.mrq_workers.find().count() > 1


def get_diff_after_jobs(worker, n_tasks, leak, sleep=0):

    time.sleep(3)

    mem_start = worker.get_report(with_memory=True)["process"]["mem"]["total"]

    worker.send_tasks(
        "tests.tasks.general.Leak",
        [{"size": leak, "sleep": sleep} for _ in range(n_tasks)],
        queue="default",
        block=True
    )

    time.sleep(3)

    mem_stop = worker.get_report(with_memory=True)["process"]["mem"]["total"]

    diff = mem_stop - mem_start

    print("Memory diff for %s tasks was %s" % (n_tasks, diff))

    return diff


def test_memoryleaks_noleak(worker):

    return pytest.skip("Too flaky, investigate sources of noise")

    TRACE = ""
    # TRACE = "--trace_memory_type ObjectId"

    worker.start(
        flags="--trace_memory --greenlets 1 --mongodb_logs 0 --scheduler_interval 0 --subqueues_refresh_interval 0 --paused_queues_refresh_interval 0 --report_interval 10000 %s" % TRACE)

    # Send it once to add to imports
    get_diff_after_jobs(worker, 10, 0)

    diff100 = get_diff_after_jobs(worker, 100, 0)

    assert worker.mongodb_jobs.mrq_jobs.count() == 100 + 10

    diff200 = get_diff_after_jobs(worker, 200, 0)

    assert worker.mongodb_jobs.mrq_jobs.count() == 200 + 100 + 10

    # Most of the tasks should have mem_diff == 0
    assert worker.mongodb_jobs.mrq_jobs.find(
        {"memory_diff": 0}).count() > 310 * 0.95

    assert diff100 < 15000
    assert diff200 < 15000

    assert worker.mongodb_jobs.mrq_workers.find().count() == 1


def test_memoryleaks_1mleak(worker):

    worker.start(
        flags="--trace_memory --greenlets 1 --mongodb_logs 0 --report_interval 10000")

    # Send it once to add to imports
    get_diff_after_jobs(worker, 10, 0)

    worker.mongodb_jobs.mrq_jobs.remove()

    # 1M leak!
    # sleep is needed so that psutil measurements are accurate :-/
    diff1m = get_diff_after_jobs(worker, 10, 100000, sleep=0)

    assert diff1m > 900000

    assert worker.mongodb_jobs.mrq_jobs.find(
        {"memory_diff": {"$gte": 80000}}).count() == 10

    assert worker.mongodb_jobs.mrq_workers.find().count() == 1