iotop/data.py

   1 import errno
   2 import glob
   3 import os
   4 import pprint
   5 import pwd
   6 import socket
   7 import stat
   8 import struct
   9 import sys
  10 import time
  11
  12 #
  13 # Check for requirements:
  14 #   o Python >= 2.5 for AF_NETLINK sockets
  15 #   o Linux >= 2.6.20 with I/O accounting
  16 #
  17 try:
  18     socket.NETLINK_ROUTE
  19     python25 = True
  20 except AttributeError:
  21     python25 = False
  22
  23 ioaccounting = os.path.exists('/proc/self/io')
  24
  25 if not python25 or not ioaccounting:
  26     def boolean2string(boolean):
  27         return boolean and 'Found' or 'Not found'
  28     print 'Could not run iotop as some of the requirements are not met:'
  29     print '- Python >= 2.5 for AF_NETLINK support:', boolean2string(python25)
  30     print '- Linux >= 2.6.20 with I/O accounting support ' \
  31              '(CONFIG_TASKSTATS, CONFIG_TASK_DELAY_ACCT, ' \
  32              'CONFIG_TASK_IO_ACCOUNTING):', \
  33           boolean2string(ioaccounting)
  34     sys.exit(1)
  35
  36 from iotop import ioprio, vmstat
  37 from netlink import Connection, NETLINK_GENERIC, U32Attr, NLM_F_REQUEST
  38 from genetlink import Controller, GeNlMessage
  39
  40 class DumpableObject(object):
  41     """Base class for all objects that allows easy introspection when printed"""
  42     def __repr__(self):
  43         return '%s: %s>' % (str(type(self))[:-1], pprint.pformat(self.__dict__))
  44
  45
  46 #
  47 # Interesting fields in a taskstats output
  48 #
  49
  50 class Stats(DumpableObject):
  51     members_offsets = [
  52         ('blkio_delay_total', 40),
  53         ('swapin_delay_total', 56),
  54         ('read_bytes', 248),
  55         ('write_bytes', 256),
  56         ('cancelled_write_bytes', 264)
  57     ]
  58
  59     has_blkio_delay_total = False
  60
  61     def __init__(self, task_stats_buffer):
  62         sd = self.__dict__
  63         for name, offset in Stats.members_offsets:
  64             data = task_stats_buffer[offset:offset + 8]
  65             sd[name] = struct.unpack('Q', data)[0]
  66
  67         # This is a heuristic to detect if CONFIG_TASK_DELAY_ACCT is enabled in
  68         # the kernel.
  69         if not Stats.has_blkio_delay_total:
  70             Stats.has_blkio_delay_total = self.blkio_delay_total != 0
  71
  72     def accumulate(self, other_stats, destination, coeff=1):
  73         """Update destination from operator(self, other_stats)"""
  74         dd = destination.__dict__
  75         sd = self.__dict__
  76         od = other_stats.__dict__
  77         for member, offset in Stats.members_offsets:
  78             dd[member] = sd[member] + coeff * od[member]
  79
  80     def delta(self, other_stats, destination):
  81         """Update destination with self - other_stats"""
  82         return self.accumulate(other_stats, destination, coeff=-1)
  83
  84     def is_all_zero(self):
  85         sd = self.__dict__
  86         for name, offset in Stats.members_offsets:
  87             if sd[name] is not 0:
  88                 return False
  89         return True
  90
  91     @staticmethod
  92     def build_all_zero():
  93         stats = Stats.__new__(Stats)
  94         std = stats.__dict__
  95         for name, offset in Stats.members_offsets:
  96             std[name] = 0
  97         return stats
  98
  99 #
 100 # Netlink usage for taskstats
 101 #
 102
 103 TASKSTATS_CMD_GET = 1
 104 TASKSTATS_CMD_ATTR_PID = 1
 105
 106 class TaskStatsNetlink(object):
 107     # Keep in sync with format_stats() and pinfo.did_some_io()
 108
 109     def __init__(self, options):
 110         self.options = options
 111         self.connection = Connection(NETLINK_GENERIC)
 112         controller = Controller(self.connection)
 113         self.family_id = controller.get_family_id('TASKSTATS')
 114
 115     def build_request(self, tid):
 116         return GeNlMessage(self.family_id, cmd=TASKSTATS_CMD_GET,
 117                            attrs=[U32Attr(TASKSTATS_CMD_ATTR_PID, tid)],
 118                            flags=NLM_F_REQUEST)
 119
 120     def get_single_task_stats(self, thread):
 121         thread.task_stats_request.send(self.connection)
 122         try:
 123             reply = self.connection.recv()
 124         except OSError, e:
 125             if e.errno == errno.ESRCH:
 126                 # OSError: Netlink error: No such process (3)
 127                 return
 128             raise
 129         if len(reply.payload) < 292:
 130             # Short reply
 131             return
 132         reply_data = reply.payload[20:]
 133
 134         reply_length, reply_type = struct.unpack('HH', reply.payload[4:8])
 135         reply_version = struct.unpack('H', reply.payload[20:22])[0]
 136         assert reply_length >= 288
 137         assert reply_type == TASKSTATS_CMD_ATTR_PID + 3
 138         assert reply_version >= 4
 139         return Stats(reply_data)
 140
 141 #
 142 # PIDs manipulations
 143 #
 144
 145 def find_uids(options):
 146     """Build options.uids from options.users by resolving usernames to UIDs"""
 147     options.uids = []
 148     error = False
 149     for u in options.users or []:
 150         try:
 151             uid = int(u)
 152         except ValueError:
 153             try:
 154                 passwd = pwd.getpwnam(u)
 155             except KeyError:
 156                 print >> sys.stderr, 'Unknown user:', u
 157                 error = True
 158             else:
 159                 uid = passwd.pw_uid
 160         if not error:
 161             options.uids.append(uid)
 162     if error:
 163         sys.exit(1)
 164
 165 def safe_utf8_decode(s):
 166     try:
 167         return s.decode('utf-8')
 168     except UnicodeDecodeError:
 169         return s.encode('string_escape')
 170
 171 class ThreadInfo(DumpableObject):
 172     """Stats for a single thread"""
 173     def __init__(self, tid, taskstats_connection):
 174         self.tid = tid
 175         self.mark = True
 176         self.stats_total = None
 177         self.stats_delta = Stats.__new__(Stats)
 178         self.task_stats_request = taskstats_connection.build_request(tid)
 179
 180     def get_ioprio(self):
 181         return ioprio.get(self.tid)
 182
 183     def update_stats(self, stats):
 184         if not self.stats_total:
 185             self.stats_total = stats
 186         stats.delta(self.stats_total, self.stats_delta)
 187         self.stats_total = stats
 188
 189
 190 class ProcessInfo(DumpableObject):
 191     """Stats for a single process (a single line in the output): if
 192     options.processes is set, it is a collection of threads, otherwise a single
 193     thread."""
 194     def __init__(self, pid):
 195         self.pid = pid
 196         self.uid = None
 197         self.user = None
 198         self.threads = {} # {tid: ThreadInfo}
 199         self.stats_delta = Stats.build_all_zero()
 200         self.stats_accum = Stats.build_all_zero()
 201         self.stats_accum_timestamp = time.time()
 202
 203     def is_monitored(self, options):
 204         if (options.pids and not options.processes and
 205             self.pid not in options.pids):
 206             # We only monitor some threads, not this one
 207             return False
 208
 209         if options.uids and self.get_uid() not in options.uids:
 210             # We only monitor some users, not this one
 211             return False
 212
 213         return True
 214
 215     def get_uid(self):
 216         if self.uid:
 217             return self.uid
 218         # uid in (None, 0) means either we don't know the UID yet or the process
 219         # runs as root so it can change its UID. In both cases it means we have
 220         # to find out its current UID.
 221         try:
 222             uid = os.stat('/proc/%d' % self.pid)[stat.ST_UID]
 223         except OSError:
 224             # The process disappeared
 225             uid = None
 226         if uid != self.uid:
 227             # Maybe the process called setuid()
 228             self.user = None
 229             self.uid = uid
 230         return uid
 231
 232     def get_user(self):
 233         uid = self.get_uid()
 234         if uid is not None and not self.user:
 235             try:
 236                 self.user = safe_utf8_decode(pwd.getpwuid(uid).pw_name)
 237             except KeyError:
 238                 self.user = str(uid)
 239         return self.user or '{none}'
 240
 241     def get_proc_status_name(self):
 242         try:
 243             proc_status = open('/proc/%d/status' % self.pid)
 244         except IOError:
 245             return '{no such process}'
 246         first_line = proc_status.readline()
 247         prefix = 'Name:\t'
 248         if first_line.startswith(prefix):
 249             name = first_line[6:].strip()
 250         else:
 251             name = ''
 252         if name:
 253             name = '[%s]' % name
 254         else:
 255             name = '{no name}'
 256         return name
 257
 258     def get_cmdline(self):
 259         # A process may exec, so we must always reread its cmdline
 260         try:
 261             proc_cmdline = open('/proc/%d/cmdline' % self.pid)
 262             cmdline = proc_cmdline.read(4096)
 263         except IOError:
 264             return '{no such process}'
 265         if not cmdline:
 266             # Probably a kernel thread, get its name from /proc/PID/status
 267             return self.get_proc_status_name()
 268         parts = cmdline.split('\0')
 269         if parts[0].startswith('/'):
 270             first_command_char = parts[0].rfind('/') + 1
 271             parts[0] = parts[0][first_command_char:]
 272         cmdline = ' '.join(parts).strip()
 273         return safe_utf8_decode(cmdline)
 274
 275     def did_some_io(self, accumulated):
 276         if accumulated:
 277             return not self.stats_accum.is_all_zero()
 278         return not all(t.stats_delta.is_all_zero() for
 279                                                  t in self.threads.itervalues())
 280
 281     def get_ioprio(self):
 282         priorities = set(t.get_ioprio() for t in self.threads.itervalues())
 283         if len(priorities) == 1:
 284             return priorities.pop()
 285         return '?dif'
 286
 287     def ioprio_sort_key(self):
 288         return ioprio.sort_key(self.get_ioprio())
 289
 290     def get_thread(self, tid, taskstats_connection):
 291         thread = self.threads.get(tid, None)
 292         if not thread:
 293             thread = ThreadInfo(tid, taskstats_connection)
 294             self.threads[tid] = thread
 295         return thread
 296
 297     def update_stats(self):
 298         stats_delta = Stats.build_all_zero()
 299         for tid, thread in self.threads.items():
 300             if thread.mark:
 301                 del self.threads[tid]
 302             else:
 303                 stats_delta.accumulate(thread.stats_delta, stats_delta)
 304
 305         nr_threads = len(self.threads)
 306         if not nr_threads:
 307             return False
 308
 309         stats_delta.blkio_delay_total /= nr_threads
 310         stats_delta.swapin_delay_total /= nr_threads
 311
 312         self.stats_delta = stats_delta
 313         self.stats_accum.accumulate(self.stats_delta, self.stats_accum)
 314
 315         return True
 316
 317 class ProcessList(DumpableObject):
 318     def __init__(self, taskstats_connection, options):
 319         # {pid: ProcessInfo}
 320         self.processes = {}
 321         self.taskstats_connection = taskstats_connection
 322         self.options = options
 323         self.timestamp = time.time()
 324         self.vmstat = vmstat.VmStat()
 325
 326         # A first time as we are interested in the delta
 327         self.update_process_counts()
 328
 329     def get_process(self, pid):
 330         """Either get the specified PID from self.processes or build a new
 331         ProcessInfo if we see this PID for the first time"""
 332         process = self.processes.get(pid, None)
 333         if not process:
 334             process = ProcessInfo(pid)
 335             self.processes[pid] = process
 336
 337         if process.is_monitored(self.options):
 338             return process
 339
 340     def list_tgids(self):
 341         if self.options.pids:
 342             return self.options.pids
 343
 344         tgids = os.listdir('/proc')
 345         if self.options.processes:
 346             return [int(tgid) for tgid in tgids if '0' <= tgid[0] <= '9']
 347
 348         tids = []
 349         for tgid in tgids:
 350             if '0' <= tgid[0] <= '9':
 351                 try:
 352                     tids.extend(map(int, os.listdir('/proc/' + tgid + '/task')))
 353                 except OSError:
 354                     # The PID went away
 355                     pass
 356         return tids
 357
 358     def list_tids(self, tgid):
 359         if not self.options.processes:
 360             return [tgid]
 361
 362         try:
 363             tids = map(int, os.listdir('/proc/%d/task' % tgid))
 364         except OSError:
 365             return []
 366
 367         if self.options.pids:
 368             tids = list(set(self.options.pids).intersection(set(tids)))
 369
 370         return tids
 371
 372     def update_process_counts(self):
 373         new_timestamp = time.time()
 374         self.duration = new_timestamp - self.timestamp
 375         self.timestamp = new_timestamp
 376
 377         for tgid in self.list_tgids():
 378             process = self.get_process(tgid)
 379             if not process:
 380                 continue
 381             for tid in self.list_tids(tgid):
 382                 thread = process.get_thread(tid, self.taskstats_connection)
 383                 stats = self.taskstats_connection.get_single_task_stats(thread)
 384                 if stats:
 385                     thread.update_stats(stats)
 386                     thread.mark = False
 387
 388         return self.vmstat.delta()
 389
 390     def refresh_processes(self):
 391         for process in self.processes.itervalues():
 392             for thread in process.threads.itervalues():
 393                 thread.mark = True
 394
 395         total_read_and_write = self.update_process_counts()
 396
 397         for pid, process in self.processes.items():
 398             if not process.update_stats():
 399                 del self.processes[pid]
 400
 401         return total_read_and_write
 402
 403     def clear(self):
 404         self.processes = {}