client.py

   1 import re, struct
   2 import git
   3 from helpers import *
   4 from subprocess import Popen, PIPE
   5
   6 class ClientError(Exception):
   7     pass
   8
   9 class Client:
  10     def __init__(self, remote, create=False):
  11         self._busy = None
  12         self._indexes_synced = 0
  13         rs = remote.split(':', 1)
  14         if len(rs) == 1:
  15             (host, dir) = ('NONE', remote)
  16             argv = ['bup', 'server']
  17         else:
  18             (host, dir) = rs
  19             argv = ['ssh', host, '--', 'bup', 'server']
  20         (self.host, self.dir) = (host, dir)
  21         self.cachedir = git.repo('index-cache/%s'
  22                                  % re.sub(r'[^@:\w]', '_',
  23                                           "%s:%s" % (host, dir)))
  24         self.p = p = Popen(argv, stdin=PIPE, stdout=PIPE)
  25         self.conn = conn = Conn(p.stdout, p.stdin)
  26         if dir:
  27             dir = re.sub(r'[\r\n]', ' ', dir)
  28             if create:
  29                 conn.write('init-dir %s\n' % dir)
  30             else:
  31                 conn.write('set-dir %s\n' % dir)
  32             conn.check_ok()
  33
  34     def __del__(self):
  35         self.close()
  36
  37     def close(self):
  38         if self.conn and not self._busy:
  39             self.conn.write('quit\n')
  40         if self.p:
  41             self.p.stdin.close()
  42             while self.p.stdout.read(65536):
  43                 pass
  44             self.p.stdout.close()
  45             self.p.wait()
  46             rv = self.p.wait()
  47             if rv:
  48                 raise ClientError('server tunnel returned exit code %d' % rv)
  49         self.conn = None
  50         self.p = None
  51
  52     def check_busy(self):
  53         if self._busy:
  54             raise ClientError('already busy with command %r' % self._busy)
  55
  56     def _not_busy(self):
  57         self._busy = None
  58
  59     def sync_indexes(self):
  60         self.check_busy()
  61         conn = self.conn
  62         conn.write('list-indexes\n')
  63         packdir = git.repo('objects/pack')
  64         mkdirp(self.cachedir)
  65         all = {}
  66         needed = {}
  67         for line in linereader(conn):
  68             if not line:
  69                 break
  70             all[line] = 1
  71             assert(line.find('/') < 0)
  72             if not os.path.exists(os.path.join(self.cachedir, line)):
  73                 needed[line] = 1
  74         conn.check_ok()
  75
  76         for f in os.listdir(self.cachedir):
  77             if f.endswith('.idx') and not f in all:
  78                 log('pruning old index: %r\n' % f)
  79                 os.unlink(os.path.join(self.cachedir, f))
  80
  81         # FIXME this should be pipelined: request multiple indexes at a time, or
  82         # we waste lots of network turnarounds.
  83         for name in needed.keys():
  84             log('requesting %r\n' % name)
  85             conn.write('send-index %s\n' % name)
  86             n = struct.unpack('!I', conn.read(4))[0]
  87             assert(n)
  88             log('   expect %d bytes\n' % n)
  89             fn = os.path.join(self.cachedir, name)
  90             f = open(fn + '.tmp', 'w')
  91             for b in chunkyreader(conn, n):
  92                 f.write(b)
  93             conn.check_ok()
  94             f.close()
  95             os.rename(fn + '.tmp', fn)
  96
  97         self._indexes_synced = 1
  98
  99     def _make_objcache(self):
 100         ob = self._busy
 101         self._busy = None
 102         self.sync_indexes()
 103         self._busy = ob
 104         return git.MultiPackIndex(self.cachedir)
 105
 106     def new_packwriter(self):
 107         self.check_busy()
 108         self._busy = 'receive-objects'
 109         return PackWriter_Remote(self.conn,
 110                                  objcache_maker = self._make_objcache,
 111                                  onclose = self._not_busy)
 112
 113     def read_ref(self, refname):
 114         self.check_busy()
 115         self.conn.write('read-ref %s\n' % refname)
 116         r = self.conn.readline().strip()
 117         self.conn.check_ok()
 118         if r:
 119             assert(len(r) == 40)   # hexified sha
 120             return r.decode('hex')
 121         else:
 122             return None   # nonexistent ref
 123
 124     def update_ref(self, refname, newval, oldval):
 125         self.check_busy()
 126         self.conn.write('update-ref %s\n%s\n%s\n'
 127                         % (refname, newval.encode('hex'),
 128                            (oldval or '').encode('hex')))
 129         self.conn.check_ok()
 130
 131     def cat(self, id):
 132         self.check_busy()
 133         self._busy = 'cat'
 134         self.conn.write('cat %s\n' % re.sub(r'[\n\r]', '_', id))
 135         while 1:
 136             sz = struct.unpack('!I', self.conn.read(4))[0]
 137             if not sz: break
 138             yield self.conn.read(sz)
 139         self.conn.check_ok()
 140         self._not_busy()
 141
 142
 143 class PackWriter_Remote(git.PackWriter):
 144     def __init__(self, conn, objcache_maker=None, onclose=None):
 145         git.PackWriter.__init__(self, objcache_maker)
 146         self.file = conn
 147         self.filename = 'remote socket'
 148         self.onclose = onclose
 149         self._packopen = False
 150
 151     def _open(self):
 152         if not self._packopen:
 153             self._make_objcache()
 154             self.file.write('receive-objects\n')
 155             self._packopen = True
 156
 157     def _end(self):
 158         if self._packopen and self.file:
 159             self.file.write('\0\0\0\0')
 160             self._packopen = False
 161             id = self.file.readline().strip()
 162             self.file.check_ok()
 163             self.objcache = None
 164             return id
 165
 166     def close(self):
 167         id = self._end()
 168         self.file = None
 169         return id
 170
 171     def abort(self):
 172         raise GitError("don't know how to abort remote pack writing")
 173
 174     def _raw_write(self, datalist):
 175         assert(self.file)
 176         if not self._packopen:
 177             self._open()
 178         data = ''.join(datalist)
 179         assert(len(data))
 180         self.file.write(struct.pack('!I', len(data)) + data)
 181         self.outbytes += len(data)
 182         self.count += 1
 183
 184