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
 201 def parse_proc_pid_status(pid):
 202     result_dict = {}
 203     try:
 204         for line in open('/proc/%d/status' % pid):
 205             key, value = line.split(':\t', 1)
 206             result_dict[key] = value.strip()
 207     except IOError:
 208         pass  # No such process
 209     return result_dict
 210
 211
 212 def safe_utf8_decode(s):
 213     try:
 214         return s.decode('utf-8')
 215     except UnicodeDecodeError:
 216         return s.encode('string_escape')
 217
 218 class ThreadInfo(DumpableObject):
 219     """Stats for a single thread"""
 220     def __init__(self, tid, taskstats_connection):
 221         self.tid = tid
 222         self.mark = True
 223         self.stats_total = None
 224         self.stats_delta = Stats.__new__(Stats)
 225         self.task_stats_request = taskstats_connection.build_request(tid)
 226
 227     def get_ioprio(self):
 228         return ioprio.get(self.tid)
 229
 230     def set_ioprio(self, ioprio_class, ioprio_data):
 231         return ioprio.set_ioprio(ioprio.IOPRIO_WHO_PROCESS, self.tid,
 232                                  ioprio_class, ioprio_data)
 233
 234     def update_stats(self, stats):
 235         if not self.stats_total:
 236             self.stats_total = stats
 237         stats.delta(self.stats_total, self.stats_delta)
 238         self.stats_total = stats
 239
 240
 241 class ProcessInfo(DumpableObject):
 242     """Stats for a single process (a single line in the output): if
 243     options.processes is set, it is a collection of threads, otherwise a single
 244     thread."""
 245     def __init__(self, pid):
 246         self.pid = pid
 247         self.uid = None
 248         self.user = None
 249         self.threads = {} # {tid: ThreadInfo}
 250         self.stats_delta = Stats.build_all_zero()
 251         self.stats_accum = Stats.build_all_zero()
 252         self.stats_accum_timestamp = time.time()
 253
 254     def is_monitored(self, options):
 255         if (options.pids and not options.processes and
 256             self.pid not in options.pids):
 257             # We only monitor some threads, not this one
 258             return False
 259
 260         if options.uids and self.get_uid() not in options.uids:
 261             # We only monitor some users, not this one
 262             return False
 263
 264         return True
 265
 266     def get_uid(self):
 267         if self.uid:
 268             return self.uid
 269         # uid in (None, 0) means either we don't know the UID yet or the process
 270         # runs as root so it can change its UID. In both cases it means we have
 271         # to find out its current UID.
 272         try:
 273             uid = os.stat('/proc/%d' % self.pid)[stat.ST_UID]
 274         except OSError:
 275             # The process disappeared
 276             uid = None
 277         if uid != self.uid:
 278             # Maybe the process called setuid()
 279             self.user = None
 280             self.uid = uid
 281         return uid
 282
 283     def get_user(self):
 284         uid = self.get_uid()
 285         if uid is not None and not self.user:
 286             try:
 287                 self.user = safe_utf8_decode(pwd.getpwuid(uid).pw_name)
 288             except KeyError:
 289                 self.user = str(uid)
 290         return self.user or '{none}'
 291
 292     def get_cmdline(self):
 293         # A process may exec, so we must always reread its cmdline
 294         try:
 295             proc_cmdline = open('/proc/%d/cmdline' % self.pid)
 296             cmdline = proc_cmdline.read(4096)
 297         except IOError:
 298             return '{no such process}'
 299         proc_status = parse_proc_pid_status(self.pid)
 300         if not cmdline:
 301             # Probably a kernel thread, get its name from /proc/PID/status
 302             proc_status_name = proc_status.get('Name', '')
 303             if proc_status_name:
 304                 proc_status_name = '[%s]' % proc_status_name
 305             else:
 306                 proc_status_name = '{no name}'
 307             return proc_status_name
 308         suffix = ''
 309         tgid = int(proc_status.get('Tgid', self.pid))
 310         if tgid != self.pid:
 311             # Not the main thread, maybe it has a custom name
 312             tgid_name = parse_proc_pid_status(tgid).get('Name', '')
 313             thread_name = proc_status.get('Name', '')
 314             if thread_name != tgid_name:
 315                 suffix += ' [%s]' % thread_name
 316         parts = cmdline.split('\0')
 317         if parts[0].startswith('/'):
 318             first_command_char = parts[0].rfind('/') + 1
 319             parts[0] = parts[0][first_command_char:]
 320         cmdline = ' '.join(parts).strip()
 321         return safe_utf8_decode(cmdline + suffix)
 322
 323     def did_some_io(self, accumulated):
 324         if accumulated:
 325             return not self.stats_accum.is_all_zero()
 326         for t in self.threads.itervalues():
 327             if not t.stats_delta.is_all_zero():
 328                 return True
 329         return False
 330
 331     def get_ioprio(self):
 332         priorities = set(t.get_ioprio() for t in self.threads.itervalues())
 333         if len(priorities) == 1:
 334             return priorities.pop()
 335         return '?dif'
 336
 337     def set_ioprio(self, ioprio_class, ioprio_data):
 338         for thread in self.threads.itervalues():
 339             thread.set_ioprio(ioprio_class, ioprio_data)
 340
 341     def ioprio_sort_key(self):
 342         return ioprio.sort_key(self.get_ioprio())
 343
 344     def get_thread(self, tid, taskstats_connection):
 345         thread = self.threads.get(tid, None)
 346         if not thread:
 347             thread = ThreadInfo(tid, taskstats_connection)
 348             self.threads[tid] = thread
 349         return thread
 350
 351     def update_stats(self):
 352         stats_delta = Stats.build_all_zero()
 353         for tid, thread in self.threads.items():
 354             if thread.mark:
 355                 del self.threads[tid]
 356             else:
 357                 stats_delta.accumulate(thread.stats_delta, stats_delta)
 358
 359         nr_threads = len(self.threads)
 360         if not nr_threads:
 361             return False
 362
 363         stats_delta.blkio_delay_total /= nr_threads
 364         stats_delta.swapin_delay_total /= nr_threads
 365
 366         self.stats_delta = stats_delta
 367         self.stats_accum.accumulate(self.stats_delta, self.stats_accum)
 368
 369         return True
 370
 371 class ProcessList(DumpableObject):
 372     def __init__(self, taskstats_connection, options):
 373         # {pid: ProcessInfo}
 374         self.processes = {}
 375         self.taskstats_connection = taskstats_connection
 376         self.options = options
 377         self.timestamp = time.time()
 378         self.vmstat = vmstat.VmStat()
 379
 380         # A first time as we are interested in the delta
 381         self.update_process_counts()
 382
 383     def get_process(self, pid):
 384         """Either get the specified PID from self.processes or build a new
 385         ProcessInfo if we see this PID for the first time"""
 386         process = self.processes.get(pid, None)
 387         if not process:
 388             process = ProcessInfo(pid)
 389             self.processes[pid] = process
 390
 391         if process.is_monitored(self.options):
 392             return process
 393
 394     def list_tgids(self):
 395         if self.options.pids:
 396             return self.options.pids
 397
 398         tgids = os.listdir('/proc')
 399         if self.options.processes:
 400             return [int(tgid) for tgid in tgids if '0' <= tgid[0] <= '9']
 401
 402         tids = []
 403         for tgid in tgids:
 404             if '0' <= tgid[0] <= '9':
 405                 try:
 406                     tids.extend(map(int, 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 = 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         for tgid in self.list_tgids():
 432             process = self.get_process(tgid)
 433             if not process:
 434                 continue
 435             for tid in self.list_tids(tgid):
 436                 thread = process.get_thread(tid, self.taskstats_connection)
 437                 stats = self.taskstats_connection.get_single_task_stats(thread)
 438                 if stats:
 439                     thread.update_stats(stats)
 440                     thread.mark = False
 441
 442         return self.vmstat.delta()
 443
 444     def refresh_processes(self):
 445         for process in self.processes.itervalues():
 446             for thread in process.threads.itervalues():
 447                 thread.mark = True
 448
 449         total_read_and_write = self.update_process_counts()
 450
 451         for pid, process in self.processes.items():
 452             if not process.update_stats():
 453                 del self.processes[pid]
 454
 455         return total_read_and_write
 456
 457     def clear(self):
 458         self.processes = {}