cmd/server-cmd.py

   1 #!/bin/sh
   2 """": # -*-python-*-
   3 bup_python="$(dirname "$0")/bup-python" || exit $?
   4 exec "$bup_python" "$0" ${1+"$@"}
   5 """
   6 # end of bup preamble
   7
   8 from __future__ import absolute_import
   9 import os, sys, struct, subprocess
  10
  11 from bup import options, git, vfs, vint
  12 from bup.compat import hexstr
  13 from bup.git import MissingObject
  14 from bup.helpers import (Conn, debug1, debug2, linereader, lines_until_sentinel,
  15                          log)
  16 from bup.repo import LocalRepo
  17
  18
  19 suspended_w = None
  20 dumb_server_mode = False
  21 repo = None
  22
  23 def do_help(conn, junk):
  24     conn.write('Commands:\n    %s\n' % '\n    '.join(sorted(commands)))
  25     conn.ok()
  26
  27
  28 def _set_mode():
  29     global dumb_server_mode
  30     dumb_server_mode = os.path.exists(git.repo('bup-dumb-server'))
  31     debug1('bup server: serving in %s mode\n'
  32            % (dumb_server_mode and 'dumb' or 'smart'))
  33
  34
  35 def _init_session(reinit_with_new_repopath=None):
  36     global repo
  37     if reinit_with_new_repopath is None and git.repodir:
  38         if not repo:
  39             repo = LocalRepo()
  40         return
  41     git.check_repo_or_die(reinit_with_new_repopath)
  42     if repo:
  43         repo.close()
  44     repo = LocalRepo()
  45     # OK. we now know the path is a proper repository. Record this path in the
  46     # environment so that subprocesses inherit it and know where to operate.
  47     os.environ['BUP_DIR'] = git.repodir
  48     debug1('bup server: bupdir is %r\n' % git.repodir)
  49     _set_mode()
  50
  51
  52 def init_dir(conn, arg):
  53     git.init_repo(arg)
  54     debug1('bup server: bupdir initialized: %r\n' % git.repodir)
  55     _init_session(arg)
  56     conn.ok()
  57
  58
  59 def set_dir(conn, arg):
  60     _init_session(arg)
  61     conn.ok()
  62
  63
  64 def list_indexes(conn, junk):
  65     _init_session()
  66     suffix = ''
  67     if dumb_server_mode:
  68         suffix = ' load'
  69     for f in os.listdir(git.repo('objects/pack')):
  70         if f.endswith('.idx'):
  71             conn.write('%s%s\n' % (f, suffix))
  72     conn.ok()
  73
  74
  75 def send_index(conn, name):
  76     _init_session()
  77     assert(name.find('/') < 0)
  78     assert(name.endswith('.idx'))
  79     idx = git.open_idx(git.repo('objects/pack/%s' % name))
  80     conn.write(struct.pack('!I', len(idx.map)))
  81     conn.write(idx.map)
  82     conn.ok()
  83
  84
  85 def receive_objects_v2(conn, junk):
  86     global suspended_w
  87     _init_session()
  88     suggested = set()
  89     if suspended_w:
  90         w = suspended_w
  91         suspended_w = None
  92     else:
  93         if dumb_server_mode:
  94             w = git.PackWriter(objcache_maker=None)
  95         else:
  96             w = git.PackWriter()
  97     while 1:
  98         ns = conn.read(4)
  99         if not ns:
 100             w.abort()
 101             raise Exception('object read: expected length header, got EOF\n')
 102         n = struct.unpack('!I', ns)[0]
 103         #debug2('expecting %d bytes\n' % n)
 104         if not n:
 105             debug1('bup server: received %d object%s.\n'
 106                 % (w.count, w.count!=1 and "s" or ''))
 107             fullpath = w.close(run_midx=not dumb_server_mode)
 108             if fullpath:
 109                 (dir, name) = os.path.split(fullpath)
 110                 conn.write('%s.idx\n' % name)
 111             conn.ok()
 112             return
 113         elif n == 0xffffffff:
 114             debug2('bup server: receive-objects suspended.\n')
 115             suspended_w = w
 116             conn.ok()
 117             return
 118
 119         shar = conn.read(20)
 120         crcr = struct.unpack('!I', conn.read(4))[0]
 121         n -= 20 + 4
 122         buf = conn.read(n)  # object sizes in bup are reasonably small
 123         #debug2('read %d bytes\n' % n)
 124         _check(w, n, len(buf), 'object read: expected %d bytes, got %d\n')
 125         if not dumb_server_mode:
 126             oldpack = w.exists(shar, want_source=True)
 127             if oldpack:
 128                 assert(not oldpack == True)
 129                 assert(oldpack.endswith('.idx'))
 130                 (dir,name) = os.path.split(oldpack)
 131                 if not (name in suggested):
 132                     debug1("bup server: suggesting index %s\n"
 133                            % git.shorten_hash(name))
 134                     debug1("bup server:   because of object %s\n"
 135                            % hexstr(shar))
 136                     conn.write('index %s\n' % name)
 137                     suggested.add(name)
 138                 continue
 139         nw, crc = w._raw_write((buf,), sha=shar)
 140         _check(w, crcr, crc, 'object read: expected crc %d, got %d\n')
 141     # NOTREACHED
 142
 143
 144 def _check(w, expected, actual, msg):
 145     if expected != actual:
 146         w.abort()
 147         raise Exception(msg % (expected, actual))
 148
 149
 150 def read_ref(conn, refname):
 151     _init_session()
 152     r = git.read_ref(refname)
 153     conn.write('%s\n' % (r or '').encode('hex'))
 154     conn.ok()
 155
 156
 157 def update_ref(conn, refname):
 158     _init_session()
 159     newval = conn.readline().strip()
 160     oldval = conn.readline().strip()
 161     git.update_ref(refname, newval.decode('hex'), oldval.decode('hex'))
 162     conn.ok()
 163
 164 def join(conn, id):
 165     _init_session()
 166     try:
 167         for blob in git.cp().join(id):
 168             conn.write(struct.pack('!I', len(blob)))
 169             conn.write(blob)
 170     except KeyError as e:
 171         log('server: error: %s\n' % e)
 172         conn.write('\0\0\0\0')
 173         conn.error(e)
 174     else:
 175         conn.write('\0\0\0\0')
 176         conn.ok()
 177
 178 def cat_batch(conn, dummy):
 179     _init_session()
 180     cat_pipe = git.cp()
 181     # For now, avoid potential deadlock by just reading them all
 182     for ref in tuple(lines_until_sentinel(conn, '\n', Exception)):
 183         ref = ref[:-1]
 184         it = cat_pipe.get(ref)
 185         info = next(it)
 186         if not info[0]:
 187             conn.write('missing\n')
 188             continue
 189         conn.write('%s %s %d\n' % info)
 190         for buf in it:
 191             conn.write(buf)
 192     conn.ok()
 193
 194 def refs(conn, args):
 195     limit_to_heads, limit_to_tags = args.split()
 196     assert limit_to_heads in ('0', '1')
 197     assert limit_to_tags in ('0', '1')
 198     limit_to_heads = int(limit_to_heads)
 199     limit_to_tags = int(limit_to_tags)
 200     _init_session()
 201     patterns = tuple(x[:-1] for x in lines_until_sentinel(conn, '\n', Exception))
 202     for name, oid in git.list_refs(patterns=patterns,
 203                                    limit_to_heads=limit_to_heads,
 204                                    limit_to_tags=limit_to_tags):
 205         assert '\n' not in name
 206         conn.write('%s %s\n' % (oid.encode('hex'), name))
 207     conn.write('\n')
 208     conn.ok()
 209
 210 def rev_list(conn, _):
 211     _init_session()
 212     count = conn.readline()
 213     if not count:
 214         raise Exception('Unexpected EOF while reading rev-list count')
 215     count = None if count == '\n' else int(count)
 216     fmt = conn.readline()
 217     if not fmt:
 218         raise Exception('Unexpected EOF while reading rev-list format')
 219     fmt = None if fmt == '\n' else fmt[:-1]
 220     refs = tuple(x[:-1] for x in lines_until_sentinel(conn, '\n', Exception))
 221     args = git.rev_list_invocation(refs, count=count, format=fmt)
 222     p = subprocess.Popen(git.rev_list_invocation(refs, count=count, format=fmt),
 223                          env=git._gitenv(git.repodir),
 224                          stdout=subprocess.PIPE)
 225     while True:
 226         out = p.stdout.read(64 * 1024)
 227         if not out:
 228             break
 229         conn.write(out)
 230     conn.write('\n')
 231     rv = p.wait()  # not fatal
 232     if rv:
 233         msg = 'git rev-list returned error %d' % rv
 234         conn.error(msg)
 235         raise GitError(msg)
 236     conn.ok()
 237
 238 def resolve(conn, args):
 239     _init_session()
 240     (flags,) = args.split()
 241     flags = int(flags)
 242     want_meta = bool(flags & 1)
 243     follow = bool(flags & 2)
 244     have_parent = bool(flags & 4)
 245     parent = vfs.read_resolution(conn) if have_parent else None
 246     path = vint.read_bvec(conn)
 247     if not len(path):
 248         raise Exception('Empty resolve path')
 249     try:
 250         res = list(vfs.resolve(repo, path, parent=parent, want_meta=want_meta,
 251                                follow=follow))
 252     except vfs.IOError as ex:
 253         res = ex
 254     if isinstance(res, vfs.IOError):
 255         conn.write(b'\0')  # error
 256         vfs.write_ioerror(conn, res)
 257     else:
 258         conn.write(b'\1')  # success
 259         vfs.write_resolution(conn, res)
 260     conn.ok()
 261
 262 optspec = """
 263 bup server
 264 """
 265 o = options.Options(optspec)
 266 (opt, flags, extra) = o.parse(sys.argv[1:])
 267
 268 if extra:
 269     o.fatal('no arguments expected')
 270
 271 debug2('bup server: reading from stdin.\n')
 272
 273 commands = {
 274     'quit': None,
 275     'help': do_help,
 276     'init-dir': init_dir,
 277     'set-dir': set_dir,
 278     'list-indexes': list_indexes,
 279     'send-index': send_index,
 280     'receive-objects-v2': receive_objects_v2,
 281     'read-ref': read_ref,
 282     'update-ref': update_ref,
 283     'join': join,
 284     'cat': join,  # apocryphal alias
 285     'cat-batch' : cat_batch,
 286     'refs': refs,
 287     'rev-list': rev_list,
 288     'resolve': resolve
 289 }
 290
 291 # FIXME: this protocol is totally lame and not at all future-proof.
 292 # (Especially since we abort completely as soon as *anything* bad happens)
 293 conn = Conn(sys.stdin, sys.stdout)
 294 lr = linereader(conn)
 295 for _line in lr:
 296     line = _line.strip()
 297     if not line:
 298         continue
 299     debug1('bup server: command: %r\n' % line)
 300     words = line.split(' ', 1)
 301     cmd = words[0]
 302     rest = len(words)>1 and words[1] or ''
 303     if cmd == 'quit':
 304         break
 305     else:
 306         cmd = commands.get(cmd)
 307         if cmd:
 308             cmd(conn, rest)
 309         else:
 310             raise Exception('unknown server command: %r\n' % line)
 311
 312 debug1('bup server: done\n')