Add RSS memory status collecting routine

Added RSS memory status collecting routine which parses file:

  /sys/fs/cgroup/memory/memory.stat

Closes tarantool/tarantool-qa#98

Author: avtikhon

dispatcher.py

@@ -41,6 +41,7 @@
 from listeners import FailWatcher
 from listeners import HangWatcher
 from listeners import LogOutputWatcher
+from listeners import RSSMonitor
 from listeners import OutputWatcher
 from listeners import StatisticsWatcher
 
@@ -121,10 +122,11 @@ def __init__(self, task_groups, max_workers_cnt, randomize):
             self.result_queues.append(task_queue_disp.result_queue)
             self.task_queues.append(task_queue_disp.task_queue)
 
-        self.report_timeout = 1.0
+        self.report_timeout = 0.1
 
         self.statistics = None
         self.artifacts = None
+        self.rss_monitor = None
         self.fail_watcher = None
         self.listeners = None
         self.init_listeners()
@@ -166,7 +168,8 @@ def init_listeners(self):
         self.statistics = StatisticsWatcher(log_output_watcher.get_logfile)
         self.artifacts = ArtifactsWatcher(log_output_watcher.get_logfile)
         output_watcher = OutputWatcher()
-        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts]
+        self.rss_monitor = RSSMonitor(output_watcher.not_done_worker_ids)
+        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts, self.rss_monitor]
         if watch_fail:
             self.fail_watcher = FailWatcher(self.terminate_all_workers)
             self.listeners.append(self.fail_watcher)

lib/utils.py

@@ -1,4 +1,5 @@
 import os
+import re
 import sys
 import collections
 import signal
@@ -233,6 +234,19 @@ def format_process(pid):
     return 'process %d [%s; %s]' % (pid, status, cmdline)
 
 
+def get_proc_stat_rss():
+    rss = 'unknown'
+    pat = re.compile(r"^total_rss .*")
+    try:
+        with open('/sys/fs/cgroup/memory/memory.stat', 'r') as f:
+            for line in f:
+                if pat.match(line) is not None:
+                    rss = int(line.split(' ')[1])
+    except (OSError, IOError):
+        pass
+    return rss
+
+
 def set_fd_cloexec(socket):
     flags = fcntl.fcntl(socket, fcntl.F_GETFD)
     fcntl.fcntl(socket, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)

listeners.py

@@ -2,6 +2,7 @@
 import sys
 import yaml
 import shutil
+import time
 
 from lib import Options
 from lib.colorer import color_stdout
@@ -15,6 +16,7 @@
 from lib.utils import safe_makedirs
 from lib.utils import print_tail_n
 from lib.utils import print_unidiff
+from lib.utils import get_proc_stat_rss
 
 
 class BaseWatcher(object):
@@ -178,6 +180,43 @@ def __del__(self):
                 pass
 
 
+class RSSMonitor(BaseWatcher):
+    def __init__(self, get_not_done_worker_ids):
+        self.get_not_done_worker_ids = get_not_done_worker_ids
+        self.rss_results = []
+        self.activity = 0.0
+        self.worker_current_task = dict()
+
+    def process_result(self, obj):
+        if isinstance(obj, WorkerCurrentTask):
+            self.worker_current_task[obj.worker_id] = obj
+
+        if not isinstance(obj, WorkerCurrentTask):
+            return
+
+        print("ALX process_result ================ {}\n" . format(obj.task_name))
+        self.rss_results.append((obj.task_name,
+                                 get_proc_stat_rss()))
+
+    def process_timeout(self, delta_seconds):
+        rss_new = []
+        self.activity += delta_seconds
+        worker_ids = self.get_not_done_worker_ids()
+
+        running_tasks = [task for worker_id, task
+                      in self.worker_current_task.items()
+                      if worker_id in worker_ids]
+        rss = get_proc_stat_rss()
+        for task_name, rss_old in self.rss_results:
+            for task in running_tasks:
+                if task_name == task.task_name:
+                    if rss_old > rss:
+                        rss = rss_old
+                    print("ALX process_timeout ============  {}: '{}' {}\n" . format(round(self.activity, 1), task_name, rss))
+            rss_new.append((task_name, rss))
+        self.rss_results = rss_new
+
+
 class OutputWatcher(BaseWatcher):
     def __init__(self):
         self.buffer = dict()