iotop/data.py

   1 # This program is free software; you can redistribute it and/or modify
   2 # it under the terms of the GNU General Public License as published by
   3 # the Free Software Foundation; either version 2 of the License, or
   4 # (at your option) any later version.
   5 #
   6 # This program is distributed in the hope that it will be useful,
   7 # but WITHOUT ANY WARRANTY; without even the implied warranty of
   8 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   9 # GNU Library General Public License for more details.
  10 #
  11 # You should have received a copy of the GNU General Public License
  12 # along with this program; if not, write to the Free Software
  13 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  14 #
  15 # See the COPYING file for license information.
  16 #
  17 # Copyright (c) 2007 Guillaume Chazarain <guichaz@gmail.com>
  18
  19 # Allow printing with same syntax in Python 2/3
  20 from __future__ import print_function
  21
  22 import errno
  23 import os
  24 import pprint
  25 import pwd
  26 import stat
  27 import struct
  28 import sys
  29 import time
  30
  31 #
  32 # Check for requirements:
  33 #   o Linux >= 2.6.20 with I/O accounting and VM event counters
  34 #
  35
  36 ioaccounting = os.path.exists('/proc/self/io')
  37
  38 try:
  39     from iotop.vmstat import VmStat
  40     vmstat_f = VmStat()
  41 except:
  42     vm_event_counters = False
  43 else:
  44     vm_event_counters = True
  45
  46 if not ioaccounting or not vm_event_counters:
  47     print('Could not run iotop as some of the requirements are not met:')
  48     print('- Linux >= 2.6.20 with')
  49     if not ioaccounting:
  50         print('  - I/O accounting support ' \
  51           '(CONFIG_TASKSTATS, CONFIG_TASK_DELAY_ACCT, ' \
  52           'CONFIG_TASK_IO_ACCOUNTING)')
  53     if not vm_event_counters:
  54         print('  - VM event counters (CONFIG_VM_EVENT_COUNTERS)')
  55     sys.exit(1)
  56
  57 from iotop import ioprio, vmstat
  58 from iotop.netlink import Connection, NETLINK_GENERIC, U32Attr, NLM_F_REQUEST
  59 from iotop.genetlink import Controller, GeNlMessage
  60
  61
  62 class DumpableObject(object):
  63     """Base class for objects that allows easy introspection when printed"""
  64     def __repr__(self):
  65         return '%s: %s>' % (str(type(self))[:-1],
  66                             pprint.pformat(self.__dict__))
  67
  68
  69 #
  70 # Interesting fields in a taskstats output
  71 #
  72
  73 class Stats(DumpableObject):
  74     members_offsets = [
  75         ('blkio_delay_total', 40),
  76         ('swapin_delay_total', 56),
  77         ('read_bytes', 248),
  78         ('write_bytes', 256),
  79         ('cancelled_write_bytes', 264)
  80     ]
  81
  82     has_blkio_delay_total = False
  83
  84     def __init__(self, task_stats_buffer):
  85         sd = self.__dict__
  86         for name, offset in Stats.members_offsets:
  87             data = task_stats_buffer[offset:offset + 8]
  88             sd[name] = struct.unpack('Q', data)[0]
  89
  90         # This is a heuristic to detect if CONFIG_TASK_DELAY_ACCT is enabled in
  91         # the kernel.
  92         if not Stats.has_blkio_delay_total:
  93             Stats.has_blkio_delay_total = self.blkio_delay_total != 0
  94
  95     def accumulate(self, other_stats, destination, coeff=1):
  96         """Update destination from operator(self, other_stats)"""
  97         dd = destination.__dict__
  98         sd = self.__dict__
  99         od = other_stats.__dict__
 100         for member, offset in Stats.members_offsets:
 101             dd[member] = sd[member] + coeff * od[member]
 102
 103     def delta(self, other_stats, destination):
 104         """Update destination with self - other_stats"""
 105         return self.accumulate(other_stats, destination, coeff=-1)
 106
 107     def is_all_zero(self):
 108         sd = self.__dict__
 109         for name, offset in Stats.members_offsets:
 110             if sd[name] != 0:
 111                 return False
 112         return True
 113
 114     @staticmethod
 115     def build_all_zero():
 116         stats = Stats.__new__(Stats)
 117         std = stats.__dict__
 118         for name, offset in Stats.members_offsets:
 119             std[name] = 0
 120         return stats
 121
 122 #
 123 # Netlink usage for taskstats
 124 #
 125
 126 TASKSTATS_CMD_GET = 1
 127 TASKSTATS_CMD_ATTR_PID = 1
 128 TASKSTATS_TYPE_AGGR_PID = 4
 129 TASKSTATS_TYPE_PID = 1
 130 TASKSTATS_TYPE_STATS = 3
 131
 132
 133 class TaskStatsNetlink(object):
 134     # Keep in sync with format_stats() and pinfo.did_some_io()
 135
 136     def __init__(self, options):
 137         self.options = options
 138         self.connection = Connection(NETLINK_GENERIC)
 139         controller = Controller(self.connection)
 140         self.family_id = controller.get_family_id('TASKSTATS')
 141
 142     def build_request(self, tid):
 143         return GeNlMessage(self.family_id, cmd=TASKSTATS_CMD_GET,
 144                            attrs=[U32Attr(TASKSTATS_CMD_ATTR_PID, tid)],
 145                            flags=NLM_F_REQUEST)
 146
 147     def get_single_task_stats(self, thread):
 148         thread.task_stats_request.send(self.connection)
 149         try:
 150             reply = GeNlMessage.recv(self.connection)
 151         except OSError as e:
 152             if e.errno == errno.ESRCH:
 153                 # OSError: Netlink error: No such process (3)
 154                 return
 155             raise
 156         for attr_type, attr_value in reply.attrs.items():
 157             if attr_type == TASKSTATS_TYPE_AGGR_PID:
 158                 reply = attr_value.nested()
 159                 break
 160         else:
 161             return
 162         taskstats_data = reply[TASKSTATS_TYPE_STATS].data
 163         if len(taskstats_data) < 272:
 164             # Short reply
 165             return
 166         taskstats_version = struct.unpack('H', taskstats_data[:2])[0]
 167         assert taskstats_version >= 4
 168         return Stats(taskstats_data)
 169
 170 #
 171 # PIDs manipulations
 172 #
 173
 174
 175 def find_uids(options):
 176     """Build options.uids from options.users by resolving usernames to UIDs"""
 177     options.uids = []
 178     error = False
 179     for u in options.users or []:
 180         try:
 181             uid = int(u)
 182         except ValueError:
 183             try:
 184                 passwd = pwd.getpwnam(u)
 185             except KeyError:
 186                 print('Unknown user:', u, file=sys.stderr)
 187                 error = True
 188             else:
 189                 uid = passwd.pw_uid
 190         if not error:
 191             options.uids.append(uid)
 192     if error:
 193         sys.exit(1)
 194
 195
 196 def parse_proc_pid_status(pid):
 197     result_dict = {}
 198     try:
 199         for line in open('/proc/%d/status' % pid):
 200             key, value = line.split(':\t', 1)
 201             result_dict[key] = value.strip()
 202     except IOError:
 203         pass  # No such process
 204     return result_dict
 205
 206
 207 def safe_utf8_decode(s):
 208     try:
 209         return s.decode('utf-8')
 210     except UnicodeDecodeError:
 211         return s.encode('string_escape')
 212     except AttributeError:
 213         return s
 214
 215
 216 class ThreadInfo(DumpableObject):
 217     """Stats for a single thread"""
 218     def __init__(self, tid, taskstats_connection):
 219         self.tid = tid
 220         self.mark = True
 221         self.stats_total = None
 222         self.stats_delta = Stats.__new__(Stats)
 223         self.task_stats_request = taskstats_connection.build_request(tid)
 224
 225     def get_ioprio(self):
 226         return ioprio.get(self.tid)
 227
 228     def set_ioprio(self, ioprio_class, ioprio_data):
 229         return ioprio.set_ioprio(ioprio.IOPRIO_WHO_PROCESS, self.tid,
 230                                  ioprio_class, ioprio_data)
 231
 232     def update_stats(self, stats):
 233         if not self.stats_total:
 234             self.stats_total = stats
 235         stats.delta(self.stats_total, self.stats_delta)
 236         self.stats_total = stats
 237
 238
 239 class ProcessInfo(DumpableObject):
 240     """Stats for a single process (a single line in the output): if
 241     options.processes is set, it is a collection of threads, otherwise a single
 242     thread."""
 243     def __init__(self, pid):
 244         self.pid = pid
 245         self.uid = None
 246         self.user = None
 247         self.threads = {}  # {tid: ThreadInfo}
 248         self.stats_delta = Stats.build_all_zero()
 249         self.stats_accum = Stats.build_all_zero()
 250         self.stats_accum_timestamp = time.time()
 251
 252     def is_monitored(self, options):
 253         if (options.pids and not options.processes and
 254             self.pid not in options.pids):
 255             # We only monitor some threads, not this one
 256             return False
 257
 258         if options.uids and self.get_uid() not in options.uids:
 259             # We only monitor some users, not this one
 260             return False
 261
 262         return True
 263
 264     def get_uid(self):
 265         if self.uid:
 266             return self.uid
 267         # uid in (None, 0) means either we don't know the UID yet or the
 268         # process runs as root so it can change its UID. In both cases it means
 269         # we have to find out its current UID.
 270         try:
 271             uid = os.stat('/proc/%d' % self.pid)[stat.ST_UID]
 272         except OSError:
 273             # The process disappeared
 274             uid = None
 275         if uid != self.uid:
 276             # Maybe the process called setuid()
 277             self.user = None
 278             self.uid = uid
 279         return uid
 280
 281     def get_user(self):
 282         uid = self.get_uid()
 283         if uid is not None and not self.user:
 284             try:
 285                 self.user = safe_utf8_decode(pwd.getpwuid(uid).pw_name)
 286             except (KeyError, AttributeError):
 287                 self.user = str(uid)
 288         return self.user or '{none}'
 289
 290     def get_cmdline(self):
 291         # A process may exec, so we must always reread its cmdline
 292         try:
 293             proc_cmdline = open('/proc/%d/cmdline' % self.pid)
 294             cmdline = proc_cmdline.read(4096)
 295         except IOError:
 296             return '{no such process}'
 297         proc_status = parse_proc_pid_status(self.pid)
 298         if not cmdline:
 299             # Probably a kernel thread, get its name from /proc/PID/status
 300             proc_status_name = proc_status.get('Name', '')
 301             if proc_status_name:
 302                 proc_status_name = '[%s]' % proc_status_name
 303             else:
 304                 proc_status_name = '{no name}'
 305             return proc_status_name
 306         suffix = ''
 307         tgid = int(proc_status.get('Tgid', self.pid))
 308         if tgid != self.pid:
 309             # Not the main thread, maybe it has a custom name
 310             tgid_name = parse_proc_pid_status(tgid).get('Name', '')
 311             thread_name = proc_status.get('Name', '')
 312             if thread_name != tgid_name:
 313                 suffix += ' [%s]' % thread_name
 314         parts = cmdline.split('\0')
 315         if parts[0].startswith('/'):
 316             first_command_char = parts[0].rfind('/') + 1
 317             parts[0] = parts[0][first_command_char:]
 318         cmdline = ' '.join(parts).strip()
 319         return safe_utf8_decode(cmdline + suffix)
 320
 321     def did_some_io(self, accumulated):
 322         if accumulated:
 323             return not self.stats_accum.is_all_zero()
 324         for t in self.threads.values():
 325             if not t.stats_delta.is_all_zero():
 326                 return True
 327         return False
 328
 329     def get_ioprio(self):
 330         priorities = set(t.get_ioprio() for t in self.threads.values())
 331         if len(priorities) == 1:
 332             return priorities.pop()
 333         return '?dif'
 334
 335     def set_ioprio(self, ioprio_class, ioprio_data):
 336         for thread in self.threads.values():
 337             thread.set_ioprio(ioprio_class, ioprio_data)
 338
 339     def ioprio_sort_key(self):
 340         return ioprio.sort_key(self.get_ioprio())
 341
 342     def get_thread(self, tid, taskstats_connection):
 343         thread = self.threads.get(tid, None)
 344         if not thread:
 345             thread = ThreadInfo(tid, taskstats_connection)
 346             self.threads[tid] = thread
 347         return thread
 348
 349     def update_stats(self):
 350         stats_delta = Stats.build_all_zero()
 351         for tid, thread in self.threads.items():
 352             if not thread.mark:
 353                 stats_delta.accumulate(thread.stats_delta, stats_delta)
 354         self.threads = dict([(tid, thread) for tid, thread in
 355                              self.threads.items() if not thread.mark])
 356
 357         nr_threads = len(self.threads)
 358         if not nr_threads:
 359             return False
 360
 361         stats_delta.blkio_delay_total /= nr_threads
 362         stats_delta.swapin_delay_total /= nr_threads
 363
 364         self.stats_delta = stats_delta
 365         self.stats_accum.accumulate(self.stats_delta, self.stats_accum)
 366
 367         return True
 368
 369
 370 class ProcessList(DumpableObject):
 371     def __init__(self, taskstats_connection, options):
 372         # {pid: ProcessInfo}
 373         self.processes = {}
 374         self.taskstats_connection = taskstats_connection
 375         self.options = options
 376         self.timestamp = time.time()
 377         self.vmstat = vmstat.VmStat()
 378
 379         # A first time as we are interested in the delta
 380         self.update_process_counts()
 381
 382     def get_process(self, pid):
 383         """Either get the specified PID from self.processes or build a new
 384         ProcessInfo if we see this PID for the first time"""
 385         process = self.processes.get(pid, None)
 386         if not process:
 387             process = ProcessInfo(pid)
 388             self.processes[pid] = process
 389
 390         if process.is_monitored(self.options):
 391             return process
 392
 393     def list_tgids(self):
 394         if self.options.pids:
 395             return self.options.pids
 396
 397         tgids = os.listdir('/proc')
 398         if self.options.processes:
 399             return [int(tgid) for tgid in tgids if '0' <= tgid[0] <= '9']
 400
 401         tids = []
 402         for tgid in tgids:
 403             if '0' <= tgid[0] <= '9':
 404                 try:
 405                     tids.extend(map(int,
 406                                     os.listdir('/proc/' + tgid + '/task')))
 407                 except OSError:
 408                     # The PID went away
 409                     pass
 410         return tids
 411
 412     def list_tids(self, tgid):
 413         if not self.options.processes:
 414             return [tgid]
 415
 416         try:
 417             tids = list(map(int, os.listdir('/proc/%d/task' % tgid)))
 418         except OSError:
 419             return []
 420
 421         if self.options.pids:
 422             tids = list(set(self.options.pids).intersection(set(tids)))
 423
 424         return tids
 425
 426     def update_process_counts(self):
 427         new_timestamp = time.time()
 428         self.duration = new_timestamp - self.timestamp
 429         self.timestamp = new_timestamp
 430
 431         total_read = total_write = 0
 432
 433         for tgid in self.list_tgids():
 434             process = self.get_process(tgid)
 435             if not process:
 436                 continue
 437             for tid in self.list_tids(tgid):
 438                 thread = process.get_thread(tid, self.taskstats_connection)
 439                 stats = self.taskstats_connection.get_single_task_stats(thread)
 440                 if stats:
 441                     thread.update_stats(stats)
 442                     delta = thread.stats_delta
 443                     total_read += delta.read_bytes
 444                     total_write += delta.write_bytes
 445                     thread.mark = False
 446         return (total_read, total_write), self.vmstat.delta()
 447
 448     def refresh_processes(self):
 449         for process in self.processes.values():
 450             for thread in process.threads.values():
 451                 thread.mark = True
 452
 453         total_read_and_write = self.update_process_counts()
 454
 455         self.processes = dict([(pid, process) for pid, process in
 456                                self.processes.items() if
 457                                process.update_stats()])
 458
 459         return total_read_and_write
 460
 461     def clear(self):
 462         self.processes = {}