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