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