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