Showing posts with label python. Show all posts
Showing posts with label python. Show all posts

Tuesday, June 30, 2009

tee functionality with python subprocess PIPEs

Sometimes we may want the output of a process to go to more then one place. In a shell, we would probably use tee(1).

One example application is dumping large databases with storage engine that can't give you a time of last modification, e.g. mysql, using innodb. Then I want the chksum of the dump, and if that is the same as the dump before, I'll unlink the older one and symlink it to the new dump. I also want to check the dump stream to make sure it has finished properly. And for good measure lets compress the stream on the fly. SURE, you could do all this after the dump completes but then you have to wait for a lot of disk I/O that the pipes avoid.

If you are like me, you'll want to use python. Note that if you take the stdout from one pipe and read it with more that one (n) other process, each process will only get a fraction of the data, ~1/n. So use a buffer to store the data and write it to each process that needs it.


__version__ = [int(x) for x in "$Revision: 1.2 $".split()[1].split('.')]
__author__ = "Kael Fischer <kael.fischer@gmail.com>"

.
.
.


syslog.syslog(syslog.LOG_ERR,"INFO: Starting mysqldump of %s on %s" % (db,host))
try:
# make output files
outFile = file(outfileName,'w')
digFile = file(digestName,'w')

# make pipes
dumper= subprocess.Popen(["mysqldump","--opt","--skip-dump-date",
"--single-transaction","--quick",db],
stdout=subprocess.PIPE)
grepper = subprocess.Popen(["grep", '-q','^-- Dump completed$'],
stdin=subprocess.PIPE,stdout=sys.stdout)
digester = subprocess.Popen(["md5", "-q"],stdin=subprocess.PIPE,stdout=digFile)
bziper = subprocess.Popen(["bzip2"],
stdin=subprocess.PIPE,stdout=outFile)

while dumper.poll() == None:
# use os.read NOT file.read
# file.read blocks.
# copy output from mysqldump to a buffer
buf=os.read(dumper.stdout.fileno(),5000)
# write buffer contents to 3 different processes
grepper.stdin.write(buf)
digester.stdin.write(buf)
bziper.stdin.write(buf)

# after dumper finishes,
# explicitly shut down input streams
# to other jobs
grepper.stdin.close()
digester.stdin.close()
bziper.stdin.close()

while grepper.poll() == None:
# wait if needed (shouldn't be)
time.sleep(1)

if grepper.returncode != 0:
raise RuntimeError("End of dump not found: grep returned - %s"%
grepper.returncode)

except BaseException, e:
# dump's no good, KeyboardInterrupt, whatever
syslog.syslog(syslog.LOG_ERR,"ERROR: %s : %s" % (type(e),e))
syslog.syslog(syslog.LOG_ERR,"ERROR: mysqldump of %s on %s failed" % (db,host))

# unsuccessful
# put files back where they were
# and exit

# exercise for reader


syslog.syslog(syslog.LOG_ERR,"INFO: mysqldump of %s on %s finished" % (db,host))

Monday, June 29, 2009

python subprocess based parallel processing

The python threading module is cool and when combined with rpyc and the Sun Grid Engine you can get a lot done really fast on a cluster. I will blog that later, but using one of the tricks I use with rpyc with the newish 'subprocess' module in the python standard library, multi-process based parallel processing seems simpler then ever now.

This is a jiffy to run a shell command on a number of hosts.



#!/usr/local/bin/python -u
#
# runAllOver.py
# Run a shell command on several machines using ssh
#
__version__ = tuple([int(x) for x in
'$Revision: 1.2 $'.split()[1].split('.')])
__author__ = "Kael Fischer"

import sys
import time
import optparse
from subprocess import Popen, PIPE


HOSTS = ["nfs1","nfs2","compute1","compute2","db1" ]

def main(sysargs):

oneLineUsage = "Usage: %prog [options] '<remote command>'"

op = optparse.OptionParser(
oneLineUsage,
version="%prog " + '.'.join([str(x) for x in __version__]))

(opts,args) = op.parse_args(sysargs)


try:
if len(args) == 0:
raise RuntimeError, "No remote command specified."
except Exception, eData:
print >> sys.stderr, ("\nUsage Error: %s\n" %eData.message)
print >> sys.stderr, op.format_help()
return 1

cmd = ' '.join(args)
print cmd

# make one running pipe object per host
pipes = [remotePipe(h,cmd) for h in HOSTS]

# report the results in turn
for i,p in enumerate(pipes):
print HOSTS[i] +':'
while p.poll() == None:
time.sleep(0.5)
print p.stdout.read()

return(0) # we did it!

def remotePipe(host,cmd,block=False):
p=Popen("ssh %s '%s'" %(host, cmd),shell=True,stdout=PIPE)
if block:
while p.poll() == None:
time.sleep(1)
return p

if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))


What's cool about that? Well all the processes are running on the hosts simultaneously and that makes it go fast.

Isn't that insecure? Could be, depending on the context. For ways to secure that kind of thing more, read this article by Brian Hatch: http://www.hackinglinuxexposed.com/articles/20021211.html. It was the basis for the intermachine communication in the PHABRIX and most especially prun was built using his authprogs as a starting concept (with greater flexibility and extra security layers added).

Friday, May 1, 2009

4.N.13

Apropos of the Sugar 4th grade math activity that David and I are working on, I wrote an integer like class that prints nicely and that allow each digit to be addressed as if it were a list.

__iter__ goes through the digits - allowing gui code to build digit boxes easily.

e.g.:

>>> print maths_4_N_13.LongDivFormatted(15670,3)
5,223
-------- R=1
3 | 15,670
>>>




from types import IntType, LongType

THOUSANDS_SEP = ','
DECIMAL_SEP = ''

class TeachingWholeNumber(object):
"""An integer whole number (>= 0) where each digit is addressable
list style. the 0th element of the 'digits' attribute
is the ones place, the 1st element is the
tens place and so on.
"""

def __init__(self,n=0):
"""New object with default value of 0
"""
if n == 0:
self.digits=[0]
else:
n=int(n)
self.digits=[]
while n > 0:
self.digits.append(n%10)
n=n/10

def __int__(self):
"""Returns the corresponding integer value
"""
retValue = 0
for i,v in enumerate(self.digits):
retValue += (10**i)*v
return retValue


#
# listy methods
#
def __len__(self):
"""Returns the length of the digits list
"""
return len(self.digits)

def __getitem__(self,i):
if type(i) not in (IntType,LongType):
return TypeError, "Index must be an integer type"
else:
return self.digits[i]

def __setitem__(self,i,v):
if type(i) not in (IntType,LongType):
return TypeError, "Index must be an integer type"
if type(v) != IntType or v > 9 or v < 0:
return ValueError, "value, v, must be an integer between 0 and 9"
if i < len(self):
self.digits[i]=v
elif v > 0:
# appending 0 past the last non-zero is a null op
while i> len(self):
self.digits.append(0)
self.digits.append(v)

#
# Expected methods for number-like type
#
def __lt__(self, other):
return int(self) < int(other)

def __le__(self, other):
return int(self) <= int(other)

def __eq__(self, other):
return int(self) == int(other)

def __ne__(self, other):
return int(self) != int(other)

def __gt__(self, other):
return int(self) > int(other)

def __ge__(self, other):
return int(self) >= int(other)

def __add__(self, other):
return self.__class__(int(self)+int(other))

def __sub__(self, other):
return self.__class__(int(self)-int(other))

def __mul__(self, other):
return self.__class__(int(self)*int(other))

def __floordiv__(self, other):
return self.__class__(int(self)/int(other))

def __mod__(self, other):
return self.__class__(int(self)%int(other))

def __divmod__(self, other):
return tuple([self.__class__(x)
for x in divmod(int(self),int(other))])

def __pow__(self, other):
return

def __lshift__(self, other):
return NotImplemented

def __rshift__(self, other):
return NotImplemented

def __and__(self, other):
return NotImplemented

def __xor__(self, other):
return NotImplemented

def __or__(self, other):
return NotImplemented

def __div__(self, other):
return self.__class__(int(self)/int(other))

def __truediv__(self, other):
return NotImplemented

def __neg__(self):
return NotImplemented

def __pos__(self):
return NotImplemented

def __abs__(self):
return abs(int(self))

def __invert__(self):
return NotImplemented

def __complex__(self):
return NotImplemented

def __long__(self):
return long(int(self))

def __float__(self):
return float(int(self))

def __repr__(self):
return str(int(self))

def __str__(self):
"""pretty string
"""

dgtList=[]
for i,v in enumerate(self):
if i>0 and (i-1)%3 == 2:
dgtList.append(THOUSANDS_SEP)
dgtList.append(str(v))
dgtList.reverse()
return ''.join(dgtList)

def rshift(self,places=1):
"""right shift (in base 10) disgarding 'places' least significant digits

Arguments:
- `self`:
- `places`:
"""
# is this better implememted by overriding >> ?

while places > 0:
self.digits.pop(0)
places -= 1

def lshift(self,places=1):
"""left shift (in base 10) disgarding 'places' least significant digits.
Arguments:
- `self`:
- `places`:
"""
# is this better implememted by overriding << ?
while places > 0:
self.digits.insert(0,0)
places -= 1

def randomTWN(low,high):
"""factory fcn to return a random TeachingWholeNumber,
between low and high inclusive.
"""

if high < 0 or low < 0:
raise ValueError, "high and low must be > 0"

i = random.randint(low,high)
return TeachingWholeNumber(i)


class Division (object):
"""Division Container
"""

def __init__ (self,dividend,divisor):
"""New one!
"""
self.dividend = dividend
self.divisor = divisor
self.quotent, self.remainder = divmod(dividend,divisor)


class LongDivFormatted(Division):
"""Knows how to print itself
"""
leftMargin = 1
rightMargin = 2
midMargin = 1
vertbar='|'
horizbar='-'

def __init__ (self,dividend,divisor,useTWN=True):
"""New one!
"""
if useTWN:
dividend=TeachingWholeNumber(dividend)
divisor=TeachingWholeNumber(divisor)
Division.__init__(self,dividend,divisor)


def __str__ (self):
"""ASCII representation.
"""
mainWidth = (self.leftMargin+
len(str(self.divisor))+
(self.midMargin *2)+
len(self.vertbar)+
len(str(self.dividend)))

remainderWidth = ((self.rightMargin) +
len(str(self.remainder))+
len('R='))

totalWidth = mainWidth+remainderWidth

retLines=[]
retLines.append(('%'+str(mainWidth)+'s')%(self.quotent))
retLines.append((' '*(self.leftMargin +
len(str(self.divisor))+
self.midMargin)) +
(self.horizbar*(len(self.vertbar)+
self.midMargin+
len(str(self.dividend)))) +
('%'+str(remainderWidth)+'s')%('R='+str(self.remainder)))
retLines.append((' ' * self.leftMargin) +
str(self.divisor) +
(' ' * self.midMargin) +
self.vertbar +
(' ' * self.midMargin) +
str(self.dividend))

return '\n'.join(retLines)

Saturday, March 28, 2009

kdbom.tryDBconnect

I put a factory function for kdbom.db instances in kdbom.kdbom (previously I had it in viroinfo).

It automatically checks localhost,3306 for the database after checking at the specified sites.... good if clients are not on the same private subnet the the server is on.



def tryDBconnect(db=None,serversPorts=None,user=None,
reuseDB=None,getTables=True,tryLocalPort=3306,
fatal=False,verboseFailure=False):
"""Try to connect to a database and return the db instance. One or more server/port
combinations can be tried. If no connection is successful a Warning is issued,
unless fatal is true in which case KdbomDatabaseError is raised.

db = databse name (string)
server = servers hostname (string)
port = server's listening port (int)

After all given (server,port) combinations are tried one or more localhost
connections over port tryLocalPort will be attempted unless that is set
to None. tryLocalPort can be an integre type or a sequence of integers.

serversPorts should be a list of (server,port) tuples, or a single 2-tuple.
serversPorts can also be a single string to make replacement of db()
calls more straight forward (defaultPort is then used for the port).

Ir reuseDB is specified, this is just a call to the db class constructor.
"""

connected = False
database = None

if reuseDB != None:
database = kdbom.db(db=db,reuseDBconnection=reuseDB,getTables=getTables)
connected = True
else:
if type (serversPorts) in StringTypes:
serversPorts = [(serversPorts,defaultPort)]
if type(serversPorts) != ListType:
if type(serversPorts) == NoneType:
serversPorts = []
elif (len(serversPorts) == 2
and type(serversPorts[0]) in StringTypes
and type(serversPorts[1]) in (IntType,LongType)):
serversPorts = [serversPorts]
elif type(serversPorts) == TupleType:
serversPorts = [serversPorts]
else:
raise ArgumentError, "serversPorts should be a list of tuples not: %s" % serversPorts
try:
for p in tryLocalPort:
serversPorts.append(('localhost',p))
except TypeError:
serversPorts.append(('localhost',tryLocalPort))

for host,port in serversPorts:
try:
print host,port
database = kdbom.db(db=db,host=host,port=port,user=user,getTables=getTables)
connected = True
except:
pass

if not connected:
if fatal:
if not verboseFailure:
raise KdbomDatabaseError, "%s DB not loaded" % db

else:
warnings.warn("%s DB not loaded" % db)
return database

Thursday, March 26, 2009

Google + Python + LLVM

There has been talk of targeting python to LLVM before (PyPy). Google is jumping in with the Unladen Swallow Project, hoping for performance multiples in the 3-5x range.

Update:
More from PyPy's author.

Thursday, March 12, 2009

finaly factored out PickleableObject

uses my nonclobbering file opener: safeOFW

class PickleableObject(object):
"""Generic pickleable class
"""
def save(self,filename,clobber=False):
"""given an object, e.g. a taxon dictionary,
pickle it into the specified filename

Set clobber = True to allow overwriting of the
output file"""
import cPickle
f = safeOFW(filename,clobber=clobber,append=False)
cPickle.dump(self,f)
f.close()

@classmethod
def restore(cls,filename):
"""given a filename, return an object containing the
pickled data in that file"""
import cPickle
f = open(filename,'rb')
obj = cPickle.load(f)
return obj



I've been meaning to move this from a class that Peter wrote it in to originally at my request. It's quite nice to have it globally available.