Changed msync to actually sync

This copies xnbd's style

.gitignore

@@ -0,0 +1,9 @@
+**/*.o
+**/*~
+flexnbd
+build/
+pkg/
+**/*.orig
+**/.*.swp
+cscope.out
+valgrind.out

.gitlab-ci.yml

@@ -0,0 +1,27 @@
+stages:
+  - package
+  - publish
+ 
+package:jessie: &package
+  stage: package
+  image: $CI_REGISTRY/docker-images/layers:$DISTRO-deb
+  variables:
+    DISTRO: jessie
+  script:
+    - package
+  artifacts:
+    paths:
+      - pkg/
+
+package:stretch:
+  <<: *package
+  variables:
+    DISTRO: stretch
+
+publish:
+  stage: publish
+  tags: 
+    - shell
+  script:
+    - publish
+

.hgignore

@@ -1,9 +0,0 @@
-.o$
-~$
-^flexnbd$
-^build/
-^pkg/
-\.orig$
-.*\.swp$
-cscope.out$
-valgrind.out$

Makefile

@@ -12,20 +12,34 @@ else
 	CFLAGS_EXTRA=-O2
 endif
 
-CCFLAGS=-D_GNU_SOURCE=1 -Wall -Wextra -Werror-implicit-function-declaration -Wstrict-prototypes -Wno-missing-field-initializers $(CFLAGS_EXTRA) $(CFLAGS)
+NO_MSYNC=1
-LLDFLAGS=-lrt -lev $(LDFLAGS_EXTRA) $(LDFLAGS)
+ifdef NO_MSYNC
+	CFLAGS_EXTRA += -DNO_MSYNC
+endif
+
+CFLAGS_EXTRA  += -fPIC --std=gnu99
+LDFLAGS_EXTRA += -Wl,--relax,--gc-sections -L$(LIB) -Wl,-rpath-link,$(LIB)
+
+# The -Wunreachable-code warning is only implemented in clang, but it
+# doesn't break anything for gcc to see it.
+WARNINGS=-Wall \
+				 -Wextra \
+				 -Werror-implicit-function-declaration \
+				 -Wstrict-prototypes \
+				 -Wno-missing-field-initializers \
+				 -Wunreachable-code
+
+CCFLAGS=-D_GNU_SOURCE=1 $(WARNINGS) $(CFLAGS_EXTRA) $(CFLAGS)
+LLDFLAGS=-lm -lrt -lev $(LDFLAGS_EXTRA) $(LDFLAGS)
 
 CC?=gcc
 
 LIBS=-lpthread
 INC=-I/usr/include/libev -Isrc/common -Isrc/server -Isrc/proxy
-COMPILE=$(CC) $(INC) -c $(CCFLAGS)
+COMPILE=$(CC) -MMD $(INC) -c $(CCFLAGS)
-SAVEDEP=$(CC) $(INC) -MM $(CCFLAGS)
 LINK=$(CC) $(LLDFLAGS) -Isrc $(LIBS)
 
-
+LIB=build/
-EXISTING_OBJS := $(wildcard build/*.o)
--include $(EXISTING_OBJS:.o=.d)
 
 COMMON_SRC  := $(wildcard src/common/*.c)
 SERVER_SRC := $(wildcard src/server/*.c)
@@ -39,12 +53,13 @@ SRCS := $(COMMON_SRC) $(SERVER_SRC) $(PROXY_SRC)
 OBJS := $(COMMON_OBJ) $(SERVER_OBJ) $(PROXY_OBJ)
 
 
-all: build/flexnbd build/flexnbd-proxy doc
+all: build doc
+
+build: server proxy
 
 build/%.o: %.c
 	mkdir -p $(dir $@)
 	$(COMPILE) $< -o $@
-	$(SAVEDEP) $< > build/$*.d
 
 objs: $(OBJS)
 
@@ -55,41 +70,42 @@ build/flexnbd-proxy: $(COMMON_OBJ) $(PROXY_OBJ) build/proxy-main.o
 	$(LINK) $^ -o $@
 
 server: build/flexnbd
+
 proxy: build/flexnbd-proxy
 
-
 CHECK_SRC := $(wildcard tests/unit/*.c)
-CHECK_OBJ := $(CHECK_SRC:tests/unit/%.c=build/tests/%.o)
+CHECK_OBJ := $(CHECK_SRC:tests/unit/%.c=build/%.o)
 # Why can't we reuse the build/%.o rule above? Not sure.
-build/tests/%.o: tests/unit/%.c
-	mkdir -p $(dir $@)
-	$(COMPILE) $< -o $@
-	$(SAVEDEP) $< > build/tests/$*.d
 
-CHECK_BINS := $(CHECK_OBJ:build/tests/%.o=build/tests/%)
+CHECK_BINS := $(CHECK_SRC:tests/unit/%.c=build/%)
-build/tests/%: build/tests/%.o $(OBJS)
+
-	$(LINK) $^ -o $@ -lcheck
+build/check_%: build/check_%.o
+	$(LINK) $^ -o $@ $(COMMON_OBJ) $(SERVER_OBJ) -lcheck -lsubunit
 
 check_objs: $(CHECK_OBJ)
 
 check_bins: $(CHECK_BINS)
-check: $(CHECK_BINS)
+
-	for bin in $^; do $$bin; done
+check: $(OBJS) $(CHECK_BINS)
+	r=true ; for bin in $(CHECK_BINS); do $$bin || r=false; done ; $$r
+
+acceptance: build
+	cd tests/acceptance && RUBYOPT='-I.' ruby nbd_scenarios -v
+
+test: check acceptance
 
 build/flexnbd.1: README.txt
-	a2x --destination-dir build --format manpage $<
+	txt2man -t flexnbd -s 1 $< > $@
+
 build/flexnbd-proxy.1: README.proxy.txt
-	a2x --destination-dir build --format manpage $<
+	txt2man -t flexnbd-proxy -s 1 $< > $@
+
 # If we don't pipe to file, gzip clobbers the original, causing make
 # to rebuild each time
 %.1.gz: %.1
 	gzip -c -f $< > $@
 
-
+doc: build/flexnbd.1.gz build/flexnbd-proxy.1.gz
-server-man: build/flexnbd.1.gz
-proxy-man: build/flexnbd-proxy.1.gz
-
-doc: server-man proxy-man
 
 install:
 	mkdir -p $(INSTALLDIR)
@@ -99,4 +115,7 @@ clean:
 	rm -rf build/*
 
 
-.PHONY: clean objs check_objs all server proxy check_bins check server-man proxy-man doc
+.PHONY: clean objs check_objs all server proxy check_bins check doc build test acceptance
+
+# Include extra dependencies at the end, NOT before 'all'
+-include $(wildcard build/*.d)

README.proxy.txt

@@ -1,19 +1,14 @@
-FLEXNBD-PROXY(1)
-================
-:doctype: manpage
-
 NAME
-----
 
  flexnbd-proxy - A simple NBD proxy
 
 SYNOPSIS
---------
 
-*flexnbd-proxy* ['OPTIONS']
+  flexnbd-proxy --addr ADDR [--port PORT] --conn-addr ADDR
+    --conn-port PORT [--bind ADDR] [--cache[=CACHE_BYTES]] 
+    [--help] [--verbose] [--quiet]
 
 DESCRIPTION
------------
 
 flexnbd-proxy is a simple NBD proxy server that implements resilient
 connection logic for the client. It connects to an upstream NBD server
@@ -25,10 +20,6 @@ of view of the client) reconnects and retransmits the request, before
 returning the response to the client.
 
 USAGE
------
-
-  $ flexnbd-proxy --addr <ADDR> [ --port <PORT> ]
-    --conn-addr <ADDR> --conn-port <PORT> [--bind <ADDR>] [option]*
 
 Proxy requests from an NBD client to an NBD server, resiliently. Only one
 client can be connected at a time, and ACLs cannot be applied to the client, as they 
@@ -57,36 +48,39 @@ Only one request may be in-flight at a time under the current architecture; that
 doesn't seem to slow things down much relative to alternative options, but may
 be changed in the future if it becomes an issue.
 
-Options
+OPTIONS
-~~~~~~~
 
-*--addr, -l ADDR*:
+  --addr, -l ADDR  
     The address to listen on. If this begins with a '/', it is assumed to be
     a UNIX domain socket to create. Otherwise, it should be an IPv4 or IPv6
     address.
-*--port, -p PORT*:
+
+  --port, -p PORT  
     The port to listen on, if --addr is not a UNIX socket. 
 
-*--conn-addr, -C ADDR*:
+  --conn-addr, -C ADDR  
     The address of the NBD server to connect to. Required.
 
-*--conn-port, -P PORT*:
+  --conn-port, -P PORT  
     The port of the NBD server to connect to. Required.
 
-*--help, -h* :
+  --cache, -c=CACHE_BYTES  
+    If given, the size in bytes of read cache to use. CACHE_BYTES
+    defaults to 4096.
+
+  --help, -h  
     Show command or global help.
 
-*--verbose, -v* :
+  --verbose, -v  
     Output all available log information to STDERR.
 
-*--quiet, -q* :
+  --quiet, -q  
     Output as little log information as possible to STDERR.
 
-
 LOGGING
--------
+
-Log output is sent to STDERR.  If --quiet is set, no output will be seen
+Log output is sent to STDERR. If --quiet is set, no output will be
-unless the program termintes abnormally.  If neither --quiet nor
+seen unless the program termintes abnormally. If neither --quiet nor
 --verbose are set, no output will be seen unless something goes wrong
 with a specific request. If --verbose is given, every available log
 message will be seen (which, for a debug build, is many). It is not an
@@ -96,32 +90,31 @@ The log line format is:
 
     <TIMESTAMP>:<LEVEL>:<PID> <THREAD> <SOURCEFILE>:<SOURCELINE>: <MSG>
 
-*TIMESTAMP*:
+<TIMESTAMP>  
   Time the log entry was made. This is expressed in terms of monotonic ms
 
-*LEVEL*:
+<LEVEL>  
   This will be one of 'D', 'I', 'W', 'E', 'F' in increasing order of
-severity.  If flexnbd is started with the --quiet flag, only 'F' will be
+  severity. If flexnbd is started with the --quiet flag, only 'F' will
-seen.  If it is started with the --verbose flag, any from 'I' upwards
+  be seen. If it is started with the --verbose flag, any from 'I'
-will be seen.  Only if you have a debug build and start it with
+  upwards will be seen. Only if you have a debug build and start it
---verbose will you see 'D' entries.
+  with --verbose will you see 'D' entries.
 
-*PID*:
+<PID>  
   This is the process ID.
 
-*THREAD*:
+<THREAD>  
-  flexnbd-proxy is currently single-threaded, so this should be the same
+  flexnbd-proxy is currently single-threaded, so this should be the
-for all lines. That may not be the case in the future.
+  same for all lines. That may not be the case in the future.
 
-*SOURCEFILE:SOURCELINE*:
+<SOURCEFILE:SOURCELINE>  
   Identifies where in the source code this log line can be found.
 
-*MSG*:
+<MSG>  
   A short message describing what's happening, how it's being done, or
-if you're very lucky *why* it's going on.
+  if you're very lucky why it's going on.
 
-Proxying
+EXAMPLES
-~~~~~~~~
 
 The main point of the proxy mode is to allow clients that would otherwise break
 when the NBD server goes away (during a migration, for instance) to see a
@@ -154,31 +147,60 @@ The proxy notices and reconnects, fulfiling any request it has in its buffer.
 The data in myfile has been moved between physical servers without the nbd
 client process having to be disturbed at all.
 
-BUGS
+READ CACHE
-----
 
-Should be reported to nick@bytemark.co.uk.
+If the --cache option is given at the command line, either without an
+argument or with an argument greater than 0, flexnbd-proxy will use a
+read-ahead cache. The cache as currently implemented doubles each read
+request size, up to a maximum of 2xCACHE_BYTES, and retains the latter
+half in a buffer. If the next read request from the client exactly
+matches the region held in the buffer, flexnbd-proxy responds from the
+cache without making a request to the server.
+
+This pattern is designed to match sequential reads, such as those
+performed by a booting virtual machine.
+
+Note: If specifying a cache size, you must use this form:
+
+  nbd-client$ flexnbd-proxy --cache=XXXX
+
+That is, the '=' is required. This is a limitation of getopt-long.
+
+If no cache size is given, a size of 4096 bytes is assumed. Caching can
+be explicitly disabled by setting a size of 0.
+
+BUGS
+
+Should be reported via GitHub.
+
+* https://github.com/BytemarkHosting/flexnbd-c/issues
 
 Current issues include:
 
-* Only old-style NBD negotiation is supported
+* only old-style NBD negotiation is supported;
-* Only one request may be in-flight at a time
+* only one request may be in-flight at a time;
-* All I/O is blocking, and signals terminate the process immediately
+* all I/O is blocking, and signals terminate the process immediately;
-* UNIX socket support is limited to the listen address
+* UNIX socket support is limited to the listen address;
-* FLUSH and TRIM commands, and the FUA flag, are not supported
+* FLUSH and TRIM commands, and the FUA flag, are not supported;
-* DISCONNECT requests do not get passed through to the NBD server
+* DISCONNECT requests do not get passed through to the NBD server;
-* No active timeout-retry of requests - we trust the kernel's idea of failure
+* no active timeout-retry of requests - we trust the kernel's idea of
+  failure.
 
 AUTHOR
-------
+
-Written by Alex Young <alex@bytemark.co.uk>.
+Originally written by Alex Young <alex@blackkettle.org>.
 Original concept and core code by Matthew Bloch <matthew@bytemark.co.uk>.
-Proxy mode written by Nick Thomas <nick@bytemark.co.uk>
+Proxy mode written by Nick Thomas <me@ur.gs>.
 
-COPYING
+The full commit history is available on GitHub.
--------
 
-Copyright (c) 2012 Bytemark Hosting Ltd. Free use of this software is
+SEE ALSO
-granted under the terms of the GNU General Public License version 3 or
+
-later.
+flexnbd(1), nbd-client(8), xnbd-server(8), xnbd-client(8)
+
+COPYRIGHT
+
+Copyright (c) 2012-2016 Bytemark Hosting Ltd. Free use of this
+software is granted under the terms of the GNU General Public License
+version 3 or later.
 

README.txt

@@ -1,17 +1,36 @@
-FLEXNBD(1)
-==========
-:doctype: manpage
-
 NAME
-----
+
 flexnbd - A fast NBD server
 
 SYNOPSIS
---------
+
-*flexnbd* 'COMMAND' ['OPTIONS']
+  flexnbd MODE [ ARGS ]
+
+  flexnbd serve --addr ADDR --port PORT --file FILE [--sock SOCK]
+    [--default-deny] [--killswitch] [global_option]* [acl_entry]*
+
+  flexnbd listen --addr ADDR --port PORT --file FILE [--sock SOCK]
+    [--default-deny] [global_option]* [acl_entry]*
+
+  flexnbd mirror --addr ADDR --port PORT --sock SOCK [--unlink]
+    [--bind BIND_ADDR] [global_option]*
+
+  flexnbd acl --sock SOCK [acl_entry]+ [global_option]*
+
+  flexnbd break --sock SOCK [global_option]*
+
+  flexnbd status --sock SOCK [global_option]*
+
+  flexnbd read --addr ADDR --port PORT --from OFFSET --size SIZE
+    [--bind BIND_ADDR] [global_option]*
+
+  flexnbd write --addr ADDR --port PORT --from OFFSET --size SIZE
+    [--bind BIND_ADDR] [global_option]*
+
+  flexnbd help [mode] [global_option]*
 
 DESCRIPTION
------------
+
 Flexnbd is a fast NBD server which supports live migration. Live
 migration is performed by writing the data to a new server. A failed
 migration will be invisible to any connected clients.
@@ -19,57 +38,56 @@ migration will be invisible to any connected clients.
 Flexnbd tries quite hard to preserve sparsity of files it is serving,
 even across migrations.
 
-COMMANDS
+SERVE MODE
---------
+
+Serve a file.
 
-serve
-~~~~~
   $ flexnbd serve --addr <ADDR> --port <PORT> --file <FILE>
-    [--sock <SOCK>] [--default-deny] [-k] [global option]* [acl entry]*
+      [--sock <SOCK>] [--default-deny] [-k] [global_option]*
+      [acl_entry]*
 
-Serve a file. If any ACL entries are given (which should be IP
+If any ACL entries are given (which should be IP
 addresses), only those clients listed will be permitted to connect.
 
 flexnbd will continue to serve until a SIGINT, SIGQUIT, or a successful
 migration.
 
-Options
+  OPTIONS
-^^^^^^^
 
-*--addr, -l ADDR*:
+  --addr, -l ADDR  
     The address to listen on. Required.
 
-*--port, -p PORT*:
+  --port, -p PORT  
     The port to listen on. Required.
 
-*--file, -f FILE*:
+  --file, -f FILE  
     The file to serve. Must already exist. Required.
 
-*--sock, -s SOCK*:
+  --sock, -s SOCK  
     Path to a control socket to open. You will need this if you want to
     migrate, get the current status, or manipulate the access control
     list.
 
-*--default-deny, -d*:
+  --default-deny, -d  
     How to interpret an empty ACL. If --default-deny is given, an
     empty ACL will let no clients connect. If it is not given, an
     empty ACL will let any client connect.
 
-*--killswitch, -k*:
+  --killswitch, -k  
     If set, we implement a 2-minute timeout on NBD requests and
     responses. If a request takes longer than that to complete,
     the client is disconnected. This is useful to keep broken
     clients from breaking migrations, among other things.
 
-listen
+LISTEN MODE
-~~~~~~
-
-  $ flexnbd listen --addr <ADDR> --port <PORT> --file <FILE>
-    [--sock <SOCK>] [--default-deny] [global option]* [acl entry]*
 
 Listen for an inbound migration, and quit with a status of 0 on
 completion.
 
+  $ flexnbd listen --addr ADDR --port PORT --file FILE
+      [--sock SOCK] [--default-deny] [global_option]* 
+      [acl_entry]*
+
 flexnbd will wait for a successful migration, and then quit. The file
 to write the inbound migration data to must already exist before you
 run 'flexnbd listen'.
@@ -81,25 +99,24 @@ to assume that a partial migration can be resumed because the
 destination has no knowledge of whether a client has made a write to
 the source in the interim.
 
-If the migration fails for a reason which the `flexnbd listen` process
+If the migration fails for a reason which the 'flexnbd listen' process
 can't fix (say, a failed local write), it will exit with an error
 status. In this case, the sender will continually retry the migration
-until it succeeds, and you will need to restart the `flexnbd listen`
+until it succeeds, and you will need to restart the 'flexnbd listen'
 process to allow that to happen.
 
-Options
+  OPTIONS
-^^^^^^^
-As for 'serve'.
 
-mirror
+As for serve.
-~~~~~~
 
-  $ flexnbd mirror --addr <ADDR> --port <PORT> --sock SOCK
+MIRROR MODE
-      [--unlink] [--bind <BIND-ADDR>] [global option]*
 
 Start a migration from the server with control socket SOCK to the server
 listening at ADDR:PORT.
 
+  $ flexnbd mirror --addr ADDR --port PORT --sock SOCK [--unlink]
+      [--bind BIND_ADDR] [global_option]*
+
 Migration can be a slow process. Rather than block the 'flexnbd mirror'
 process until it completes, it will exit with a message of "Migration
 started" once it has confirmation that the local server was able to
@@ -112,7 +129,7 @@ again.  It is not safe to resume the migration from where it left off
 because the source can't see that the backing store behind the
 destination is intact, or even on the same machine.
 
-If the `--unlink` option is given, the local file will be deleted
+If the --unlink option is given, the local file will be deleted
 immediately before the mirror connection is terminated. This allows
 an otherwise-ambiguous situation to be resolved: if you don't unlink
 the file and the flexnbd process at either end is terminated, it's not
@@ -122,161 +139,150 @@ the data, there can be no ambiguity.
 
 Note: files smaller than 4096 bytes cannot be mirrored.
 
-Options
+  OPTIONS
-^^^^^^^
 
-*--addr, -l ADDR*:
+  --addr, -l ADDR  
     The address of the remote server to migrate to. Required.
 
-*--port, -p PORT*:
+  --port, -p PORT  
     The port of the remote server to migrate to. Required.
 
-*--sock, -s SOCK*:
+  --sock, -s SOCK  
     The control socket of the local server to migrate from. Required.
 
-*--unlink, -u*:
+  --unlink, -u  
-  Unlink the served file from the local filesystem after successfully
+    Unlink the served file from the local filesystem after
-  mirroring.
+    successfully mirroring.
 
-*--bind, -b BIND-ADDR*:
+  --bind, -b BIND_ADDR  
-  The local address to bind to. You may need this if the remote server
+    The local address to bind to. You may need this if the remote
-  is using an access control list.
+    server is using an access control list.
 
-break
+BREAK MODE
-~~~~~
-
-  $ flexnbd mirror --sock SOCK [global option]*
 
 Stop a running migration.
 
-Options
+  $ flexnbd break --sock SOCK [global_option]*
-^^^^^^^
 
-*--sock, -s SOCK*:
+  OPTIONS
-  The control socket of the local server whose emigration to stop.
+
+  --sock, -s SOCK  
+    The control socket of the local server whose migration to stop.
     Required.
 
-
+ACL MODE
-acl
-~~~
-
-  $ flexnbd acl --sock <SOCK> [acl entry]+ [global option]*
 
 Set the access control list of the server with the control socket SOCK
 to the given access control list entries.
 
+  $ flexnbd acl --sock SOCK [acl_entry]+ [global_option]*
+
 ACL entries are given as IP addresses.
 
-Options
+  OPTIONS
-^^^^^^^
 
-*--sock, -s SOCK*:
+  --sock, -s SOCK  
-  The control socket of the server whose ACL to replace.
+    The control socket of the server whose ACL to replace. Required
 
-status
+STATUS MODE
-~~~~~~
-
-  $ flexnbd status --sock <SOCK> [global option]*
 
 Get the current status of the server with control socket SOCK.
 
+  $ flexnbd status --sock SOCK [global_option]*
+
 The status will be printed to STDOUT. It is a space-separated list of
 key=value pairs. The space character will never appear in a key or
 value. Currently reported values are:
 
-*pid*:
+pid  
   The process id of the server listening on SOCK.
 
-*is_mirroring*:
+is_mirroring  
   'true' if this server is sending migration data, 'false' otherwise.
 
-*has_control*:
+has_control  
   'false' if this server was started in 'listen' mode. 'true' otherwise.
 
-read
+  OPTIONS
-~~~~
 
-  $ flexnbd read --addr <ADDR> --port <PORT> --from <OFFSET>
+  --sock, -s SOCK  
-    --size <SIZE>  [--bind BIND-ADDR] [global option]*
+    The control socket of the server of interest. Required.
+
+READ MODE
 
 Connect to the server at ADDR:PORT, and read SIZE bytes starting at
-OFFSET in a single NBD query.  The returned data will be echoed to
+OFFSET in a single NBD query.
-STDOUT.  In case of a remote ACL, set the local source address to
-BIND-ADDR.
 
-Options
+  $ flexnbd read --addr ADDR --port PORT --from OFFSET --size SIZE
-^^^^^^^
+      [--bind BIND_ADDR] [global_option]*
 
-*--addr, -l ADDR*:
+The returned data will be echoed to STDOUT. In case of a remote ACL,
+set the local source address to BIND_ADDR.
+
+  OPTIONS
+
+  --addr, -l ADDR  
     The address of the remote server. Required.
 
-*--port, -p PORT*:
+  --port, -p PORT  
     The port of the remote server. Required.
 
-*--from, -F OFFSET*:
+  --from, -F OFFSET  
     The byte offset to start reading from. Required.  Maximum 2^62.
 
-*--size, -S SIZE*:
+  --size, -S SIZE  
     The number of bytes to read. Required. Maximum 2^30.
 
-*--bind, -b BIND-ADDR*:
+  --bind, -b BIND_ADDR  
-  The local address to bind to. You may need this if the remote server
+    The local address to bind to. You may need this if the remote
-  is using an access control list.
+    server is using an access control list.
 
-write
+WRITE MODE
-~~~~~
-
-  $ cat ... | flexnbd write --addr <ADDR> --port <PORT> --from <OFFSET>
-    --size <SIZE>  [--bind BIND-ADDR] [global option]*
 
 Connect to the server at ADDR:PORT, and write SIZE bytes from STDIN
-starting at OFFSET in a single NBD query.  In case of a remote ACL, set
+starting at OFFSET in a single NBD query.
-the local source address to BIND-ADDR.
 
-Options
+  $ cat ... | flexnbd write --addr ADDR --port PORT --from OFFSET
-^^^^^^^
+      --size SIZE  [--bind BIND_ADDR] [global_option]*
 
-*--addr, -l ADDR*:
+In case of a remote ACL, set the local source address to BIND_ADDR.
+
+  OPTIONS
+
+  --addr, -l ADDR  
     The address of the remote server. Required.
 
-*--port, -p PORT*:
+  --port, -p PORT  
     The port of the remote server. Required.
 
-*--from, -F OFFSET*:
+  --from, -F OFFSET  
     The byte offset to start writing from. Required.  Maximum 2^62.
 
-*--size, -S SIZE*:
+  --size, -S SIZE  
     The number of bytes to write. Required. Maximum 2^30.
 
-*--bind, -b BIND-ADDR*:
+  --bind, -b BIND_ADDR  
-  The local address to bind to. You may need this if the remote server
+    The local address to bind to. You may need this if the remote
-  is using an access control list.
+    server is using an access control list.
 
-help
+HELP MODE
-~~~~
 
-  $ flexnbd help [command] [global option]*
+  $ flexnbd help [mode] [global_option]*
 
-Without 'command', show the list of available commands.  With 'command',
+Without mode, show the list of available modes. With mode, show help for that mode.
-show help for that command.
 
 GLOBAL OPTIONS
---------------
 
-*--help, -h* :
+--help, -h  Show mode or global help.
-    Show command or global help.
 
-*--verbose, -v* :
+--verbose, -v  Output all available log information to STDERR.
-    Output all available log information to STDERR.
-
-*--quiet, -q* :
-    Output as little log information as possible to STDERR.
 
+--quiet, -q  Output as little log information as possible to STDERR.
 
 LOGGING
--------
+
-Log output is sent to STDERR.  If --quiet is set, no output will be seen
+Log output is sent to STDERR. If --quiet is set, no output will be
-unless the program termintes abnormally.  If neither --quiet nor
+seen unless the program termintes abnormally. If neither --quiet nor
 --verbose are set, no output will be seen unless something goes wrong
 with a specific request. If --verbose is given, every available log
 message will be seen (which, for a debug build, is many). It is not an
@@ -284,39 +290,38 @@ error to set both --verbose and --quiet.  The last one wins.
 
 The log line format is:
 
-  <TIMESTAMP>:<LEVEL>:<PID> <THREAD> <SOURCEFILE>:<SOURCELINE>: <MSG>
+  <TIMESTAMP>:<LEVEL>:<PID> <THREAD> <SOURCEFILE:SOURCELINE>: <MSG>
 
-*TIMESTAMP*:
+<TIMESTAMP>  
-  Time the log entry was made. This is expressed in terms of monotonic ms.
+  Time the log entry was made. This is expressed in terms of monotonic
+  ms.
 
-*LEVEL*:
+<LEVEL>  
   This will be one of 'D', 'I', 'W', 'E', 'F' in increasing order of
-severity.  If flexnbd is started with the --quiet flag, only 'F' will be
+  severity. If flexnbd is started with the --quiet flag, only 'F'
-seen.  If it is started with the --verbose flag, any from 'I' upwards
+  will be seen. If it is started with the --verbose flag, any from 'I'
-will be seen.  Only if you have a debug build and start it with
+  upwards will be seen. Only if you have a debug build and start it
---verbose will you see 'D' entries.
+  with --verbose will you see 'D' entries.
 
-*PID*:
+<PID>  
   This is the process ID.
 
-*THREAD*:
+<THREAD>  
-  There are several pthreads per flexnbd process: a main thread, a serve
+  There are several pthreads per flexnbd process: a main thread, a
-thread, a thread per client, and possibly a pair of mirror threads and a
+  serve thread, a thread per client, and possibly a pair of mirror
-control thread.  This field identifies which thread was responsible for
+  threads and a control thread. This field identifies which thread was
-the log line.
+  responsible for the log line.
 
-*SOURCEFILE:SOURCELINE*:
+<SOURCEFILE:SOURCELINE>  
   Identifies where in the source code this log line can be found.
 
-*MSG*:
+<MSG>  
   A short message describing what's happening, how it's being done, or
-if you're very lucky *why* it's going on.
+  if you're very lucky why it's going on.
 
 EXAMPLES
---------
 
-Serving a file
+  SERVING A FILE
-~~~~~~~~~~~~~~
 
 The simplest case is serving a file on the default nbd port:
 
@@ -326,8 +331,7 @@ The simplest case is serving a file on the default nbd port:
   root:x:
   $
 
-Reading server status
+  READING SERVER STATUS
-~~~~~~~~~~~~~~~~~~~~~
 
 In order to read a server's status, we need it to open a control socket.
 
@@ -340,8 +344,7 @@ In order to read a server's status, we need it to open a control socket.
 
 Note that the status output is newline-terminated.
 
-Migrating
+  MIGRATING
-~~~~~~~~~
 
 To migrate, we need to provide a destination file of the right size.
 
@@ -367,8 +370,8 @@ With this knowledge in hand, we can start the migration:
   $ flexnbd mirror --addr 127.0.0.1 --port 4779 \
       --sock /tmp/flex-source.sock
   Migration started
-  [1]  + 9648 done       build/flexnbd serve --addr 0.0.0.0 --port 4778
+  [1]  + 9648 done       flexnbd serve --addr 0.0.0.0 --port 4778
-  [2]  + 9651 done       build/flexnbd listen --addr 0.0.0.0 --port 4779
+  [2]  + 9651 done       flexnbd listen --addr 0.0.0.0 --port 4779
   $
 
 Note that because the file is so small in this case, we see the source
@@ -376,21 +379,25 @@ server quit soon after we start the migration, and the destination
 exited at roughly the same time.
 
 BUGS
-----
 
-Should be reported to alex@bytemark.co.uk.
+Should be reported on GitHub at
+
+ * https://github.com/BytemarkHosting/flexnbd-c/issues
 
 AUTHOR
-------
 
-Written by Alex Young <alex@bytemark.co.uk>.
+Originally written by Alex Young <alex@blackkettle.org>.
 Original concept and core code by Matthew Bloch <matthew@bytemark.co.uk>.
-Some additions by Nick Thomas <nick@bytemark.co.uk>
+Proxy mode written by Nick Thomas <me@ur.gs>.
 
-COPYING
+The full commit history is available on GitHub.
--------
 
-Copyright (c) 2012 Bytemark Hosting Ltd. Free use of this software is
+SEE ALSO
-granted under the terms of the GNU General Public License version 3 or
-later.
 
+flexnbd-proxy(1), nbd-client(8), xnbd-server(8), xnbd-client(8)
+
+COPYRIGHT
+
+Copyright (c) 2012-2016 Bytemark Hosting Ltd. Free use of this
+software is granted under the terms of the GNU General Public License
+version 3 or later.

Rakefile

@@ -1,52 +0,0 @@
-# encoding: utf-8
-
-def make(*targets)
-  sh "make #{targets.map{|t| t.to_s}.join(" ")}"
-end
-
-def maketask( opts )
-  case opts
-  when Symbol
-    maketask opts => opts
-  else
-    opts.each do |name, targets|
-      task( name ){make *[*targets]}
-    end
-  end
-end
-
-
-desc "Build the binary and man page"
-maketask :build => [:all, :doc]
-
-desc "Build just the flexnbd binary"
-maketask :flexnbd => [:server]
-file "build/flexnbd" => :flexnbd
-
-desc "Build just the flexnbd-proxy binary"
-maketask :flexnbd_proxy => [:proxy]
-file "build/flexnbd-proxy" => :flexnbd_proxy
-
-desc "Build just the man page"
-maketask :man => :doc
-
-
-namespace "test" do
-  desc "Run all tests"
-  task 'run' => ["unit", "scenarios"]
-
-  desc "Build C tests"
-  maketask :build => :check_bins
-
-  desc "Run C tests"
-  maketask :unit => :check
-
-  desc "Run NBD test scenarios"
-  task 'scenarios' => ["build/flexnbd", "build/flexnbd-proxy"] do
-    sh "cd tests/acceptance && RUBYOPT='-I.' ruby nbd_scenarios -v"
-  end
-end
-
-
-desc "Remove all build targets, binaries and temporary files"
-maketask :clean

debian/control

@@ -1,14 +1,14 @@
 Source: flexnbd
-Section: unknown
+Section: web
 Priority: extra
-Maintainer: Alex Young <alex@bytemark.co.uk>
+Maintainer: Patrick J Cherry <patrick@bytemark.co.uk> 
-Build-Depends: cdbs, debhelper (>= 7.0.50), ruby, rake, gcc, libev-dev
+Build-Depends: debhelper (>= 7.0.50), ruby, gcc, libev-dev, txt2man, check, net-tools, libsubunit-dev, ruby-test-unit
 Standards-Version: 3.8.1
-Homepage: http://bigv.io/
+Homepage: https://github.com/BytemarkHosting/flexnbd-c
 
 Package: flexnbd
 Architecture: any
-Depends: ${shlibs:Depends}, ${misc:Depends}, libev3
+Depends: ${shlibs:Depends}, ${misc:Depends}, libev4 | libev3
 Description: FlexNBD server
   An NBD server offering push-mirroring and intelligent sparse file handling
 

debian/flexnbd.install

@@ -1,5 +1,3 @@
 build/flexnbd             usr/bin
 build/flexnbd-proxy       usr/bin
-build/flexnbd.1.gz        usr/share/man/man1
-build/flexnbd-proxy.1.gz  usr/share/man/man1
 

debian/flexnbd.manpages

@@ -0,0 +1,2 @@
+build/flexnbd.1.gz
+build/flexnbd-proxy.1.gz

debian/rules

@@ -7,6 +7,13 @@
 %:
 	dh $@ 
 
-.PHONY: override_dh_strip
 override_dh_strip:
 	dh_strip --dbg-package=flexnbd-dbg
+
+#
+# TODO: The ruby test suites don't work during buiding in a chroot, so leave
+# them out for now.
+#
+#override_dh_auto_test:
+#	rake test:run
+

src/common/ioutil.c

@@ -31,8 +31,6 @@ int build_allocation_map(struct bitset * allocation_map, int fd)
 
 	for (offset = 0; offset < allocation_map->size; ) {
 
-		unsigned int i;
-
 		fiemap->fm_start = offset;
 
 		fiemap->fm_length = max_length;
@@ -49,7 +47,7 @@ int build_allocation_map(struct bitset * allocation_map, int fd)
 			return 0; /* it's up to the caller to free the map */
 		}
 		else {
-			for ( i = 0; i < fiemap->fm_mapped_extents; i++ ) {
+			for ( unsigned int i = 0; i < fiemap->fm_mapped_extents; i++ ) {
 				bitset_set_range( allocation_map,
 						  fiemap->fm_extents[i].fe_logical,
 						  fiemap->fm_extents[i].fe_length );
@@ -75,18 +73,18 @@ int build_allocation_map(struct bitset * allocation_map, int fd)
 	return 1;
 }
 
-
+int open_and_mmap(const char* filename, int* out_fd, uint64_t *out_size, void **out_map)
-int open_and_mmap(const char* filename, int* out_fd, off64_t *out_size, void **out_map)
 {
+	/* 
+	 * size and out_size are intentionally of different types.
+	 * lseek64() uses off64_t to signal errors in the sign bit.
+	 * Since we check for these errors before trying to assign to
+	 * *out_size, we know *out_size can never go negative.
+	 */
 	off64_t size;
 
-	/* O_DIRECT seems to be intermittently supported.  Leaving it as
+	/* O_DIRECT should not be used with mmap() */
-	 * a compile-time option for now. */
-#ifdef DIRECT_IO
-	*out_fd = open(filename, O_RDWR | O_DIRECT | O_SYNC );
-#else
 	*out_fd = open(filename, O_RDWR | O_SYNC );
-#endif
 
 	if (*out_fd < 1) {
 		warn("open(%s) failed: does it exist?", filename);
@@ -102,15 +100,18 @@ int open_and_mmap(const char* filename, int* out_fd, off64_t *out_size, void **o
 		*out_size = size;
 	}
 
-	if (out_map) {
+	if (0) {
 		*out_map = mmap64(NULL, size, PROT_READ|PROT_WRITE, MAP_SHARED,
 		  *out_fd, 0);
 		if (((long) *out_map) == -1) {
 			warn("mmap64() failed");
 			return -1;
 		}
-	}
 		debug("opened %s size %ld on fd %d @ %p", filename, size, *out_fd, *out_map);
+	}
+	else {
+		debug("opened %s size %ld on fd %d", filename, size, *out_fd);
+	}
 
 	return 0;
 }
@@ -347,3 +348,36 @@ ssize_t iobuf_write( int fd, struct iobuf *iobuf )
 
 	return count;
 }
+
+struct iommap *iommap_alloc(int fd, off64_t from, uint64_t len) {
+
+	off66_t mmap_from = from & ~((off64_t) getpagesize() - 1);
+        uint64_t mmap_len = len + (from - mmap_from);
+        void *mmap_buf = NULL;
+
+	// TODO: Check the error code from mmap64
+	if(mmap_len)
+		mmap_buf = mmap64(NULL, mmap_len, PROT_READ | PROT_WRITE, MAP_SHARED, fd, mmap_from);
+
+	struct iommap *im = xmalloc(sizeof(struct iommap ));
+
+	im->mmap_buf = mmap_buf;
+	im->mmap_len = mmap_len;
+	im->buf = (char *) mmap_buf + from - mmap_from;
+	debug("mmapped file from %ld:%d", mmap_from, mmap_len);
+	return im;
+}
+
+void iommap_sync(struct iommap *im) {
+        if (im->mmap_len)
+		msync(im->mmap_buf, im->mmap_len, MS_SYNC | MS_INVALIDATE);
+	return;
+}
+
+void iommap_free(struct iommap *im) {
+        if (im->mmap_len)
+		munmap(im->mmap_buf, im->mmap_len);
+
+	free(im);
+}
+

src/common/ioutil.h

@@ -11,7 +11,20 @@ struct iobuf {
 ssize_t iobuf_read( int fd, struct iobuf* iobuf, size_t default_size );
 ssize_t iobuf_write( int fd, struct iobuf* iobuf );
 
+#include <inttypes.h>
+
+struct iommap {
+	char *buf;
+	char *mmap_buf;
+	uint64_t mmap_len;
+};
+
+struct iommap *iommap_alloc(int fd, off64_t from, uint64_t len);
+void   iommap_sync(struct iommap *im);
+void   iommap_free(struct iommap *im);
+
 #include "serve.h"
+
 struct bitset; /* don't need whole of bitset.h here */
 
 /** Scan the file opened in ''fd'', set bits in ''allocation_map''  that
@@ -65,7 +78,7 @@ int read_lines_until_blankline(int fd, int max_line_length, char ***lines);
   * ''out_size'' and the address of the mmap in ''out_map''.  If anything goes
   * wrong, returns -1 setting errno, otherwise 0.
   */
-int open_and_mmap( const char* filename, int* out_fd, off64_t *out_size, void **out_map);
+int open_and_mmap( const char* filename, int* out_fd, uint64_t* out_size, void **out_map);
 
 
 /** Check to see whether the given file descriptor is closed.

src/common/mode.h

@@ -7,8 +7,9 @@ void mode(char* mode, int argc, char **argv);
 
 #include <getopt.h>
 
-#define GETOPT_ARG(x,s)  {(x), 1, 0, (s)}
+#define GETOPT_ARG(x,s)    {(x), required_argument, 0, (s)}
-#define GETOPT_FLAG(x,v) {(x), 0, 0, (v)}
+#define GETOPT_FLAG(x,v)   {(x), no_argument, 0, (v)}
+#define GETOPT_OPTARG(x,s) {(x), optional_argument, 0, (s)}
 
 #define OPT_HELP "help"
 #define OPT_ADDR "addr"
@@ -19,6 +20,7 @@ void mode(char* mode, int argc, char **argv);
 #define OPT_FROM "from"
 #define OPT_SIZE "size"
 #define OPT_DENY "default-deny"
+#define OPT_CACHE "cache"
 #define OPT_UNLINK "unlink"
 #define OPT_CONNECT_ADDR "conn-addr"
 #define OPT_CONNECT_PORT "conn-port"
@@ -52,6 +54,7 @@ void mode(char* mode, int argc, char **argv);
 #define GETOPT_FROM GETOPT_ARG( OPT_FROM, 'F' )
 #define GETOPT_SIZE GETOPT_ARG( OPT_SIZE, 'S' )
 #define GETOPT_BIND GETOPT_ARG( OPT_BIND, 'b' )
+#define GETOPT_CACHE GETOPT_OPTARG( OPT_CACHE, 'c' )
 #define GETOPT_UNLINK GETOPT_ARG( OPT_UNLINK, 'u' )
 #define GETOPT_CONNECT_ADDR GETOPT_ARG( OPT_CONNECT_ADDR, 'C' )
 #define GETOPT_CONNECT_PORT GETOPT_ARG( OPT_CONNECT_PORT, 'P' )

src/common/nbdtypes.c

@@ -27,7 +27,7 @@ void nbd_r2h_request( struct nbd_request_raw *from, struct nbd_request * to )
 {
 	to->magic = htobe32( from->magic );
 	to->type = htobe32( from->type );
-	memcpy( to->handle, from->handle, 8 );
+	to->handle.w = from->handle.w;
 	to->from = htobe64( from->from );
 	to->len = htobe32( from->len );
 }
@@ -36,7 +36,7 @@ void nbd_h2r_request( struct nbd_request * from, struct nbd_request_raw * to )
 {
 	to->magic = be32toh( from->magic );
 	to->type = be32toh( from->type );
-	memcpy( to->handle, from->handle, 8 );
+	to->handle.w = from->handle.w;
 	to->from = be64toh( from->from );
 	to->len = be32toh( from->len );
 }
@@ -46,13 +46,13 @@ void nbd_r2h_reply( struct nbd_reply_raw * from, struct nbd_reply * to )
 {
 	to->magic = htobe32( from->magic );
 	to->error = htobe32( from->error );
-	memcpy( to->handle, from->handle, 8 );
+	to->handle.w = from->handle.w;
 }
 
 void nbd_h2r_reply( struct nbd_reply * from, struct nbd_reply_raw * to )
 {
 	to->magic = be32toh( from->magic );
 	to->error = be32toh( from->error );
-	memcpy( to->handle, from->handle, 8 );
+	to->handle.w = from->handle.w;
 }
 

src/common/nbdtypes.h

@@ -16,7 +16,7 @@
 
 
 /* 1MiB is the de-facto standard for maximum size of header + data */
-#define NBD_MAX_SIZE ( 1024 * 1024 )
+#define NBD_MAX_SIZE ( 32 * 1024 * 1024 )
 
 #define NBD_REQUEST_SIZE ( sizeof( struct nbd_request_raw ) )
 #define NBD_REPLY_SIZE   ( sizeof( struct nbd_reply_raw ) )
@@ -24,6 +24,11 @@
 #include <linux/types.h>
 #include <inttypes.h>
 
+typedef union nbd_handle_t {
+	uint8_t b[8];
+	uint64_t w;
+} nbd_handle_t;
+
 /* The _raw types are the types as they appear on the wire.  Non-_raw
  * types are in host-format.
  * Conversion functions are _r2h_ for converting raw to host, and _h2r_
@@ -39,7 +44,7 @@ struct nbd_init_raw {
 struct nbd_request_raw {
 	__be32 magic;
 	__be32 type;    /* == READ || == WRITE  */
-	char handle[8];
+	nbd_handle_t handle;
 	__be64 from;
 	__be32 len;
 } __attribute__((packed));
@@ -47,7 +52,7 @@ struct nbd_request_raw {
 struct nbd_reply_raw {
 	__be32 magic;
 	__be32 error;           /* 0 = ok, else error   */
-	char handle[8];         /* handle you got from request  */
+	nbd_handle_t handle;         /* handle you got from request  */
 };
 
 
@@ -62,7 +67,7 @@ struct nbd_init {
 struct nbd_request {
 	uint32_t magic;
 	uint32_t type;    /* == READ || == WRITE || == DISCONNECT */
-	char handle[8];
+	nbd_handle_t handle;
 	uint64_t from;
 	uint32_t len;
 } __attribute__((packed));
@@ -70,7 +75,7 @@ struct nbd_request {
 struct nbd_reply {
 	uint32_t magic;
 	uint32_t error;           /* 0 = ok, else error   */
-	char handle[8];         /* handle you got from request  */
+	nbd_handle_t handle;      /* handle you got from request  */
 };
 
 void nbd_r2h_init( struct nbd_init_raw * from, struct nbd_init * to );

src/common/readwrite.c

@@ -41,7 +41,7 @@ int socket_connect(struct sockaddr* to, struct sockaddr* from)
 	return fd;
 }
 
-int nbd_check_hello( struct nbd_init_raw* init_raw, off64_t* out_size )
+int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size )
 {
 	if ( strncmp( init_raw->passwd, INIT_PASSWD, 8 ) != 0 ) {
 		warn( "wrong passwd" );
@@ -62,7 +62,7 @@ fail:
 
 }
 
-int socket_nbd_read_hello( int fd, off64_t* out_size )
+int socket_nbd_read_hello( int fd, uint64_t* out_size )
 {
 	struct nbd_init_raw init_raw;
 
@@ -101,12 +101,11 @@ int socket_nbd_write_hello(int fd, off64_t out_size)
 	return 1;
 }
 
-void fill_request(struct nbd_request *request, int type, off64_t from, int len)
+void fill_request(struct nbd_request *request, int type, uint64_t from, uint32_t len)
 {
 	request->magic  = htobe32(REQUEST_MAGIC);
 	request->type   = htobe32(type);
-	((int*) request->handle)[0] = rand();
+	request->handle.w = (((uint64_t)rand()) << 32) | ((uint64_t)rand());
-	((int*) request->handle)[1] = rand();
 	request->from   = htobe64(from);
 	request->len    = htobe32(len);
 }
@@ -126,7 +125,7 @@ void read_reply(int fd, struct nbd_request *request, struct nbd_reply *reply)
 	if (reply->error != 0) {
 		error("Server replied with error %d", reply->error);
 	}
-	if (strncmp(request->handle, reply->handle, 8) != 0) {
+	if (request->handle.w != reply->handle.w) {
 		error("Did not reply with correct handle");
 	}
 }
@@ -149,7 +148,7 @@ void wait_for_data( int fd, int timeout_secs )
 }
 
 
-void socket_nbd_read(int fd, off64_t from, int len, int out_fd, void* out_buf, int timeout_secs)
+void socket_nbd_read(int fd, uint64_t from, uint32_t len, int out_fd, void* out_buf, int timeout_secs)
 {
 	struct nbd_request request;
 	struct nbd_reply   reply;
@@ -173,7 +172,7 @@ void socket_nbd_read(int fd, off64_t from, int len, int out_fd, void* out_buf, i
 	}
 }
 
-void socket_nbd_write(int fd, off64_t from, int len, int in_fd, void* in_buf, int timeout_secs)
+void socket_nbd_write(int fd, uint64_t from, uint32_t len, int in_fd, void* in_buf, int timeout_secs)
 {
 	struct nbd_request request;
 	struct nbd_reply   reply;
@@ -213,10 +212,12 @@ int socket_nbd_disconnect( int fd )
 }
 
 #define CHECK_RANGE(error_type) { \
-	off64_t size;\
+	uint64_t size;\
 	int success = socket_nbd_read_hello(params->client, &size); \
 	if ( success ) {\
-		if (params->from < 0 || (params->from + params->len) > size) {\
+		uint64_t endpoint = params->from + params->len; \
+		if (endpoint > size || \
+				endpoint < params->from ) { /* this happens on overflow */ \
 			fatal(error_type \
 			  " request %d+%d is out of range given size %d", \
 			  params->from, params->len, size\

src/common/readwrite.h

@@ -7,17 +7,17 @@
 #include "nbdtypes.h"
 
 int socket_connect(struct sockaddr* to, struct sockaddr* from);
-int socket_nbd_read_hello(int fd, off64_t * size);
+int socket_nbd_read_hello(int fd, uint64_t* size);
-int socket_nbd_write_hello(int fd, off64_t size);
+int socket_nbd_write_hello(int fd, uint64_t size);
-void socket_nbd_read(int fd, off64_t from, int len, int out_fd, void* out_buf, int timeout_secs);
+void socket_nbd_read(int fd, uint64_t from, uint32_t len, int out_fd, void* out_buf, int timeout_secs);
-void socket_nbd_write(int fd, off64_t from, int len, int out_fd, void* out_buf, int timeout_secs);
+void socket_nbd_write(int fd, uint64_t from, uint32_t len, int out_fd, void* out_buf, int timeout_secs);
 int socket_nbd_disconnect( int fd );
 
 /* as you can see, we're slowly accumulating code that should really be in an
  * NBD library */
 
-void nbd_hello_to_buf( struct nbd_init_raw* buf, off64_t out_size );
+void nbd_hello_to_buf( struct nbd_init_raw* buf, uint64_t out_size );
-int nbd_check_hello( struct nbd_init_raw* init_raw, off64_t* out_size );
+int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size );
 
 #endif
 

src/common/remote.c

@@ -63,7 +63,5 @@ void do_remote_command(char* command, char* socket_name, int argc, char** argv)
 	print_response( response );
 
 	exit(atoi(response));
-
-	close(remote);
 }
 

src/common/self_pipe.c

@@ -51,7 +51,6 @@ struct self_pipe * self_pipe_create(void)
 {
 	struct self_pipe *sig = xmalloc( sizeof( struct self_pipe ) );
 	int fds[2];
-	int fcntl_err;
 	
 	if ( NULL == sig ) { return NULL; }
 
@@ -62,7 +61,7 @@ struct self_pipe * self_pipe_create(void)
 	}
 
 	if ( fcntl( fds[0], F_SETFL, O_NONBLOCK ) || fcntl( fds[1], F_SETFL, O_NONBLOCK ) ) {
-		fcntl_err = errno;
+		int fcntl_err = errno;
 		while( close( fds[0] ) == -1 && errno == EINTR );
 		while( close( fds[1] ) == -1 && errno == EINTR );
 		free( sig );

src/common/sockutil.c

@@ -39,7 +39,6 @@ const char* sockaddr_address_string( const struct sockaddr* sa, char* dest, size
 	struct sockaddr_un*  un  = ( struct sockaddr_un*  ) sa;
 
 	unsigned short real_port = ntohs( in->sin_port ); // common to in and in6
-	size_t size;
 	const char* ret = NULL;
 
 	memset( dest, 0, len );
@@ -57,7 +56,7 @@ const char* sockaddr_address_string( const struct sockaddr* sa, char* dest, size
 	}
 
 	if ( NULL != ret && real_port > 0 && sa->sa_family != AF_UNIX ) {
-		size = strlen( dest );
+		size_t size = strlen( dest );
 		snprintf( dest + size, len - size, " port %d", real_port );
 	}
 
@@ -69,12 +68,48 @@ int sock_set_reuseaddr( int fd, int optval )
 	return setsockopt( fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval) );
 }
 
+int sock_set_keepalive_params( int fd, int time, int intvl, int probes)
+{
+	if (sock_set_keepalive(fd, 1) ||
+		sock_set_tcp_keepidle(fd, time) ||
+		sock_set_tcp_keepintvl(fd, intvl) ||
+		sock_set_tcp_keepcnt(fd, probes)) {
+		return -1;
+	}
+	return 0;
+}
+
+int sock_set_keepalive( int fd, int optval )
+{
+	return setsockopt( fd, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval) );
+}
+
+int sock_set_tcp_keepidle( int fd, int optval )
+{
+	return setsockopt( fd, IPPROTO_TCP, TCP_KEEPIDLE, &optval, sizeof(optval) );
+}
+
+int sock_set_tcp_keepintvl( int fd, int optval )
+{
+	return setsockopt( fd, IPPROTO_TCP, TCP_KEEPINTVL, &optval, sizeof(optval) );
+}
+
+int sock_set_tcp_keepcnt( int fd, int optval )
+{
+	return setsockopt( fd, IPPROTO_TCP, TCP_KEEPCNT, &optval, sizeof(optval) );
+}
+
 /* Set the tcp_nodelay option */
 int sock_set_tcp_nodelay( int fd, int optval )
 {
 	return setsockopt( fd, IPPROTO_TCP, TCP_NODELAY, &optval, sizeof(optval) );
 }
 
+int sock_set_tcp_cork( int fd, int optval )
+{
+	return setsockopt( fd, IPPROTO_TCP, TCP_CORK, &optval, sizeof(optval) );
+}
+
 int sock_set_nonblock( int fd, int optval )
 {
 	int flags = fcntl( fd, F_GETFL );
@@ -96,7 +131,7 @@ int sock_try_bind( int fd, const struct sockaddr* sa )
 {
 	int bind_result;
 	char s_address[256];
-	int retry = 1;
+	int retry = 10;
 
 	sockaddr_address_string( sa, &s_address[0], 256 );
 
@@ -122,8 +157,11 @@ int sock_try_bind( int fd, const struct sockaddr* sa )
 				 * will cope with it.
 				 */
 				case EADDRNOTAVAIL:
+					retry--;
+					if (retry) {
 						debug( "retrying" );
 						sleep( 1 );
+					}
 					continue;
 				case EADDRINUSE:
 					warn( "%s in use, giving up.", s_address );

src/common/sockutil.h

@@ -14,14 +14,29 @@ size_t sockaddr_size(const struct sockaddr* sa);
  */
 const char* sockaddr_address_string(const struct sockaddr* sa, char* dest, size_t len);
 
+/* Configure TCP keepalive on a socket */
+int sock_set_keepalive_params( int fd, int time, int intvl, int probes);
+
+/* Set the SOL_KEEPALIVE otion */
+int sock_set_keepalive(int fd, int optval);
+
 /* Set the SOL_REUSEADDR otion */
 int sock_set_reuseaddr(int fd, int optval);
 
+/* Set the tcp_keepidle option */
+int sock_set_tcp_keepidle(int fd, int optval);
+
+/* Set the tcp_keepintvl option */
+int sock_set_tcp_keepintvl(int fd, int optval);
+
+/* Set the tcp_keepcnt option */
+int sock_set_tcp_keepcnt(int fd, int optval);
+
 /* Set the tcp_nodelay option */
 int sock_set_tcp_nodelay(int fd, int optval);
 
-/* TODO: Set the tcp_cork option */
+/* Set the tcp_cork option */
-// int sock_set_cork(int fd, int optval);
+int sock_set_tcp_cork(int fd, int optval);
 
 int sock_set_nonblock(int fd, int optval);
 

src/common/util.c

@@ -13,6 +13,7 @@
 pthread_key_t cleanup_handler_key;
 
 int log_level = 2;
+char *log_context = "";
 
 void error_init(void)
 {

src/common/util.h

@@ -21,6 +21,9 @@ extern int log_level;
 /* set up the error globals */
 void error_init(void);
 
+/* some context for the overall process that appears on each log line */
+extern char *log_context;
+
 
 void exit_err( const char * );
 
@@ -92,7 +95,7 @@ uint64_t monotonic_time_ms(void);
 
 #define levstr(i) (i==0?'D':(i==1?'I':(i==2?'W':(i==3?'E':'F'))))
 
-#define myloglev(level, msg, ...) mylog( level, "%"PRIu64":%c:%d %p %s:%d: "msg"\n", monotonic_time_ms(), levstr(level), getpid(),pthread_self(), __FILE__, __LINE__, ##__VA_ARGS__ )
+#define myloglev(level, msg, ...) mylog( level, "%"PRIu64":%c:%d %p %s %s:%d: "msg"\n", monotonic_time_ms(), levstr(level), getpid(),pthread_self(), log_context, __FILE__, __LINE__, ##__VA_ARGS__ )
 
 #ifdef DEBUG
 #  define debug(msg, ...) myloglev(0, msg, ##__VA_ARGS__)
@@ -116,6 +119,7 @@ uint64_t monotonic_time_ms(void);
 #define fatal(msg, ...) do { \
 	myloglev(4, msg, ##__VA_ARGS__); \
 	error_handler(1); \
+	exit(1); /* never-reached, this is to make static code analizer happy */ \
 } while(0)
 
 

src/main.c

@@ -2,12 +2,16 @@
 #include "mode.h"
 
 #include <signal.h>
+#include <stdlib.h>
+#include <time.h>
 
 int main(int argc, char** argv)
 {
 	signal(SIGPIPE, SIG_IGN); /* calls to splice() unhelpfully throw this */
 	error_init();
 
+	srand(time(NULL));
+
 	if (argc < 2) {
 		exit_err( help_help_text );
 	}

src/proxy-main.c

@@ -1,4 +1,6 @@
 #include <signal.h>
+#include <stdlib.h>
+#include <time.h>
 
 #include "mode.h"
 #include "util.h"
@@ -12,6 +14,7 @@ static struct option proxy_options[] = {
 	GETOPT_CONNECT_ADDR,
 	GETOPT_CONNECT_PORT,
 	GETOPT_BIND,
+	GETOPT_CACHE,
 	GETOPT_QUIET,
 	GETOPT_VERBOSE,
 	{0}
@@ -27,22 +30,25 @@ static char proxy_help_text[] =
 	"\t--" OPT_CONNECT_ADDR ",-C <ADDR>\tAddress of the proxied server.\n"
 	"\t--" OPT_CONNECT_PORT ",-P <PORT>\tPort of the proxied server.\n"
 	"\t--" OPT_BIND ",-b <ADDR>\tThe address we connect from, as a proxy.\n"
+	"\t--" OPT_CACHE ",-c[=<CACHE-BYTES>]\tUse a RAM read cache of the given size.\n"
 	QUIET_LINE
 	VERBOSE_LINE;
 
+static char proxy_default_cache_size[] = "4096";
+
 void read_proxy_param(
 			int c,
 			char **downstream_addr,
 			char **downstream_port,
 			char **upstream_addr,
 			char **upstream_port,
-			char **bind_addr )
+			char **bind_addr,
+			char **cache_bytes)
 {
 	switch( c ) {
 		case 'h' :
 			fprintf( stdout, "%s\n", proxy_help_text );
 			exit( 0 );
-			break;
 		case 'l':
 			*downstream_addr = optarg;
 			break;
@@ -58,6 +64,9 @@ void read_proxy_param(
 		case 'b':
 			*bind_addr = optarg;
 			break;
+		case 'c':
+			*cache_bytes = optarg ? optarg : proxy_default_cache_size;
+			break;
 		case 'q':
 			log_level = QUIET_LOG_LEVEL;
 			break;
@@ -89,6 +98,7 @@ int main( int argc, char *argv[] )
 	char *upstream_addr   = NULL;
 	char *upstream_port   = NULL;
 	char *bind_addr       = NULL;
+	char *cache_bytes     = NULL;
 	int success;
 
 	sigset_t mask;
@@ -103,6 +113,8 @@ int main( int argc, char *argv[] )
 	exit_action.sa_mask = mask;
 	exit_action.sa_flags = 0;
 
+	srand(time(NULL));
+
 	while (1) {
 		c = getopt_long( argc, argv, proxy_short_options, proxy_options, NULL );
 		if ( -1 == c ) { break; }
@@ -111,7 +123,8 @@ int main( int argc, char *argv[] )
 				&downstream_port,
 				&upstream_addr,
 				&upstream_port,
-				&bind_addr
+				&bind_addr,
+				&cache_bytes
 		);
 	}
 
@@ -128,7 +141,8 @@ int main( int argc, char *argv[] )
 		downstream_port,
 		upstream_addr,
 		upstream_port,
-		bind_addr
+		bind_addr,
+		cache_bytes
 	);
 
 	/* Set these *after* proxy has been assigned to */

src/proxy/prefetch.c

@@ -0,0 +1,68 @@
+#include "prefetch.h"
+#include "util.h"
+
+
+struct prefetch* prefetch_create( size_t size_bytes ){
+
+	struct prefetch* out = xmalloc( sizeof( struct prefetch ) );
+	NULLCHECK( out );
+	
+	out->buffer = xmalloc( size_bytes );
+	NULLCHECK( out->buffer );
+
+	out->size = size_bytes;
+	out->is_full = 0;
+	out->from = 0;
+	out->len = 0;
+
+	return out;
+
+}
+
+void prefetch_destroy( struct prefetch *prefetch ) {
+	if( prefetch ) {
+		free( prefetch->buffer );
+		free( prefetch );
+	}
+}
+
+size_t prefetch_size( struct prefetch *prefetch){
+	if ( prefetch ) {
+		return prefetch->size;
+	} else {
+		return 0;
+	}
+}
+
+void prefetch_set_is_empty( struct prefetch *prefetch ){
+	prefetch_set_full( prefetch, 0 );
+}
+
+void prefetch_set_is_full( struct prefetch *prefetch ){
+	prefetch_set_full( prefetch, 1 );
+}
+
+void prefetch_set_full( struct prefetch *prefetch, int val ){
+	if( prefetch ) {
+		prefetch->is_full = val;
+	}
+}
+
+int prefetch_is_full( struct prefetch *prefetch ){
+	if( prefetch ) {
+		return prefetch->is_full;
+	} else {
+		return 0;
+	}
+}
+
+int prefetch_contains( struct prefetch *prefetch, uint64_t from, uint32_t len ){
+	NULLCHECK( prefetch );
+	return from >= prefetch->from &&
+		from + len <= prefetch->from + prefetch->len;
+}
+
+char *prefetch_offset( struct prefetch *prefetch, uint64_t from ){
+	NULLCHECK( prefetch );
+	return prefetch->buffer + (from - prefetch->from);
+}

src/proxy/prefetch.h

@@ -1,14 +1,33 @@
 #ifndef PREFETCH_H
 #define PREFETCH_H
 
+#include <stdint.h>
+#include <stddef.h>
+
 #define PREFETCH_BUFSIZE 4096
 
 struct prefetch {
+	/* True if there is data in the buffer. */
         int is_full;
-        __be64 from;
+	/* The start point of the current content of buffer */
-        __be32 len;
+        uint64_t from;
+	/* The length of the current content of buffer */
+        uint32_t len;
 
-        char buffer[PREFETCH_BUFSIZE];
+	/* The total size of the buffer, in bytes. */
+	size_t size;
+
+        char *buffer;
 };
 
+struct prefetch* prefetch_create( size_t size_bytes );
+void prefetch_destroy( struct prefetch *prefetch );
+size_t prefetch_size( struct prefetch *);
+void prefetch_set_is_empty( struct prefetch *prefetch );
+void prefetch_set_is_full( struct prefetch *prefetch );
+void prefetch_set_full( struct prefetch *prefetch, int val );
+int prefetch_is_full( struct prefetch *prefetch );
+int prefetch_contains( struct prefetch *prefetch, uint64_t from, uint32_t len );
+char *prefetch_offset( struct prefetch *prefetch, uint64_t from );
+
 #endif

src/proxy/proxy.c

@@ -1,9 +1,7 @@
 #include "proxy.h"
 #include "readwrite.h"
 
-#ifdef PREFETCH
 #include "prefetch.h"
-#endif
 
 
 #include "ioutil.h"
@@ -20,7 +18,8 @@ struct proxier* proxy_create(
 	char* s_downstream_port,
 	char* s_upstream_address,
 	char* s_upstream_port,
-	char* s_upstream_bind )
+	char* s_upstream_bind,
+	char* s_cache_bytes )
 {
 	struct proxier* out;
 	out = xmalloc( sizeof( struct proxier ) );
@@ -65,31 +64,49 @@ struct proxier* proxy_create(
 	out->downstream_fd = -1;
 	out->upstream_fd = -1;
 
-#ifdef PREFETCH
+	out->prefetch = NULL;
-	out->prefetch = xmalloc( sizeof( struct prefetch ) );
+	if ( s_cache_bytes ){
-#endif
+		int cache_bytes = atoi( s_cache_bytes );
+		/* leaving this off or setting a cache size of zero or
+		 * less results in no cache.
+		 */
+		if ( cache_bytes >= 0 ) {
+			out->prefetch = prefetch_create( cache_bytes );
+		}
+	}
 
 	out->init.buf = xmalloc( sizeof( struct nbd_init_raw ) );
 	out->req.buf  = xmalloc( NBD_MAX_SIZE );
 	out->rsp.buf  = xmalloc( NBD_MAX_SIZE );
 
+	log_context = xmalloc( strlen(s_upstream_address) + strlen(s_upstream_port) + 2 );
+	sprintf(log_context, "%s:%s", s_upstream_address, s_upstream_port);
+
 	return out;
 }
 
+int proxy_prefetches( struct proxier* proxy ) {
+	NULLCHECK( proxy );
+	return proxy->prefetch != NULL;
+}
+
+int proxy_prefetch_bufsize( struct proxier* proxy ){
+	NULLCHECK( proxy );
+	return prefetch_size( proxy->prefetch );
+}
+
 void proxy_destroy( struct proxier* proxy )
 {
 	free( proxy->init.buf );
 	free( proxy->req.buf );
 	free( proxy->rsp.buf );
-#ifdef PREFETCH
+	prefetch_destroy( proxy->prefetch );
-	free( proxy->prefetch );
-#endif
 
 	free( proxy );
 }
 
 /* Shared between our two different connect_to_upstream paths */
-void proxy_finish_connect_to_upstream( struct proxier *proxy, off64_t size );
+void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size );
 
 /* Try to establish a connection to our upstream server. Return 1 on success,
  * 0 on failure. this is a blocking call that returns a non-blocking socket.
@@ -102,7 +119,7 @@ int proxy_connect_to_upstream( struct proxier* proxy )
 	}
 
 	int fd = socket_connect( &proxy->connect_to.generic, connect_from );
-	off64_t size = 0;
+	uint64_t size = 0;
 
 	if ( -1 == fd ) {
 		return 0;
@@ -174,7 +191,7 @@ error:
 	return;
 }
 
-void proxy_finish_connect_to_upstream( struct proxier *proxy, off64_t size ) {
+void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size ) {
 
 	if ( proxy->upstream_size == 0 ) {
 		info( "Size of upstream image is %"PRIu64" bytes", size );
@@ -186,6 +203,13 @@ void proxy_finish_connect_to_upstream( struct proxier *proxy, off64_t size ) {
 	}
 
 	proxy->upstream_size = size;
+
+	if ( AF_UNIX != proxy->connect_to.family ) {
+		if ( sock_set_tcp_nodelay( proxy->upstream_fd, 1 ) == -1 ) {
+			warn( SHOW_ERRNO( "Failed to set TCP_NODELAY" ) );
+		}
+	}
+
 	info( "Connected to upstream on fd %i", proxy->upstream_fd );
 
 	return;
@@ -272,10 +296,9 @@ static inline int proxy_state_upstream( int state )
 	       state == WRITE_TO_UPSTREAM   || state == READ_FROM_UPSTREAM;
 }
 
-#ifdef PREFETCH
-
 int proxy_prefetch_for_request( struct proxier* proxy, int state )
 {
+	NULLCHECK( proxy );
 	struct nbd_request* req = &proxy->req_hdr;
 	struct nbd_reply* rsp = &proxy->rsp_hdr;
 
@@ -284,23 +307,11 @@ int proxy_prefetch_for_request( struct proxier* proxy, int state )
 
 	int is_read = ( req->type & REQUEST_MASK ) == REQUEST_READ;
 
-	int prefetch_start = req->from;
-	int prefetch_end = req->from + ( req->len * 2 );
-
-	/* We only want to consider prefetching if we know we're not
-	 * getting too much data back, if it's a read request, and if
-	 * the prefetch won't try to read past the end of the file.
-	 */
-
-	int prefetching = req->len <= PREFETCH_BUFSIZE && is_read &&
-		prefetch_start < prefetch_end && prefetch_end <= proxy->upstream_size;
-
 	if ( is_read ) {
 		/* See if we can respond with what's in our prefetch
 		 * cache */
-		if ( proxy->prefetch->is_full &&
+		if ( prefetch_is_full( proxy->prefetch ) &&
-		    req->from == proxy->prefetch->from &&
+			prefetch_contains( proxy->prefetch, req->from, req->len ) ) {
-		    req->len  == proxy->prefetch->len ) {
 			/* HUZZAH!  A match! */
 			debug( "Prefetch hit!" );
 
@@ -315,10 +326,11 @@ int proxy_prefetch_for_request( struct proxier* proxy, int state )
 			/* and the data */
 			memcpy(
 				proxy->rsp.buf + NBD_REPLY_SIZE,
-				proxy->prefetch->buffer, proxy->prefetch->len
+				prefetch_offset( proxy->prefetch, req->from ),
+				req->len
 			);
 
-			proxy->rsp.size = NBD_REPLY_SIZE + proxy->prefetch->len;
+			proxy->rsp.size = NBD_REPLY_SIZE + req->len;
 			proxy->rsp.needle = 0;
 
 			/* return early, our work here is done */
@@ -332,11 +344,24 @@ int proxy_prefetch_for_request( struct proxier* proxy, int state )
 		 * whether we can keep it or not.
 		 */
 		debug( "Blowing away prefetch cache on type %d request.", req->type );
-		proxy->prefetch->is_full = 0;
+		prefetch_set_is_empty( proxy->prefetch );
 	}
 
 	debug( "Prefetch cache MISS!");
 
+	uint64_t prefetch_start = req->from;
+	/* We prefetch what we expect to be the next request. */
+	uint64_t prefetch_end = req->from + ( req->len * 2 );
+
+	/* We only want to consider prefetching if we know we're not
+	 * getting too much data back, if it's a read request, and if
+	 * the prefetch won't try to read past the end of the file.
+	 */
+	int prefetching = 
+		req->len <= prefetch_size( proxy->prefetch ) && 
+		is_read &&
+		prefetch_start < prefetch_end && 
+		prefetch_end <= proxy->upstream_size; 
 
 	/* We pull the request out of the proxy struct, rewrite the
 	 * request size, and write it back.
@@ -347,7 +372,8 @@ int proxy_prefetch_for_request( struct proxier* proxy, int state )
 
 		req->len *= 2;
 
-		debug( "Prefetching %"PRIu32" bytes", req->len - proxy->prefetch_req_orig_len );
+		debug( "Prefetching additional %"PRIu32" bytes", 
+				req->len - proxy->prefetch_req_orig_len );
 		nbd_h2r_request( req, req_raw );
 	}
 
@@ -364,10 +390,10 @@ int proxy_prefetch_for_reply( struct proxier* proxy, int state )
 
 	prefetched_bytes = proxy->req_hdr.len - proxy->prefetch_req_orig_len;
 
-	debug( "Prefetched %d bytes", prefetched_bytes );
+	debug( "Prefetched additional %d bytes", prefetched_bytes );
 	memcpy(
-		proxy->rsp.buf + proxy->prefetch_req_orig_len,
+		proxy->prefetch->buffer,
-		&(proxy->prefetch->buffer),
+		proxy->rsp.buf + proxy->prefetch_req_orig_len + NBD_REPLY_SIZE,
 		prefetched_bytes
 	);
 
@@ -382,13 +408,12 @@ int proxy_prefetch_for_reply( struct proxier* proxy, int state )
 	proxy->rsp.size -= prefetched_bytes;
 
 	/* And we need to reset these */
-	proxy->prefetch->is_full = 1;
+	prefetch_set_is_full( proxy->prefetch );
 	proxy->is_prefetch_req = 0;
 
 	return state;
 }
 
-#endif
 
 
 int proxy_read_from_downstream( struct proxier *proxy, int state )
@@ -469,10 +494,8 @@ int proxy_continue_connecting_to_upstream( struct proxier* proxy, int state )
 		return state;
 	}
 
-#ifdef PREFETCH
 	/* Data may have changed while we were disconnected */
-	proxy->prefetch->is_full = 0;
+	prefetch_set_is_empty( proxy->prefetch );
-#endif
 
 	info( "Connected to upstream on fd %i", proxy->upstream_fd );
 	return READ_INIT_FROM_UPSTREAM;
@@ -492,7 +515,7 @@ int proxy_read_init_from_upstream( struct proxier* proxy, int state )
 	}
 
 	if ( proxy->init.needle == proxy->init.size ) {
-		off64_t upstream_size;
+		uint64_t upstream_size;
 		if ( !nbd_check_hello( (struct nbd_init_raw*) proxy->init.buf, &upstream_size ) ) {
 			warn( "Upstream sent invalid init" );
 			goto disconnect;
@@ -518,11 +541,22 @@ int proxy_write_to_upstream( struct proxier* proxy, int state )
 	ssize_t count;
 
 //	assert( state == WRITE_TO_UPSTREAM );
+
+	/* FIXME: We may set cork=1 multiple times as a result of this idiom.
+	 * Not a serious problem, but we could do better
+	 */
+	if ( proxy->req.needle == 0 && AF_UNIX != proxy->connect_to.family ) {
+		if ( sock_set_tcp_cork( proxy->upstream_fd, 1 ) == -1 ) {
+			warn( SHOW_ERRNO( "Failed to set TCP_CORK" ) );
+		}
+	}
+
 	count = iobuf_write( proxy->upstream_fd, &proxy->req );
 
 	if ( count == -1 ) {
 		warn( SHOW_ERRNO( "Failed to send request to upstream" ) );
 		proxy->req.needle = 0;
+		// We're throwing the socket away so no need to uncork
 		return CONNECT_TO_UPSTREAM;
 	}
 
@@ -531,6 +565,14 @@ int proxy_write_to_upstream( struct proxier* proxy, int state )
 		 * still need req.size if reading the reply fails - we disconnect
 		 * and resend the reply in that case - so keep it around for now. */
 		proxy->req.needle = 0;
+
+		if ( AF_UNIX != proxy->connect_to.family ) {
+			if ( sock_set_tcp_cork( proxy->upstream_fd, 0 ) == -1 ) {
+				warn( SHOW_ERRNO( "Failed to unset TCP_CORK" ) );
+				// TODO: should we return to CONNECT_TO_UPSTREAM in this instance?
+			}
+		}
+
 		return READ_FROM_UPSTREAM;
 	}
 
@@ -670,7 +712,7 @@ void proxy_session( struct proxier* proxy )
 			state_started = monotonic_time_ms();
 
 			debug(
-				"State transitition from %s to %s",
+				"State transition from %s to %s",
 				proxy_session_state_names[old_state],
 				proxy_session_state_names[state]
 			);
@@ -736,14 +778,12 @@ void proxy_session( struct proxier* proxy )
 			case READ_FROM_DOWNSTREAM:
 				if ( FD_ISSET( proxy->downstream_fd, &rfds ) ) {
 					state = proxy_read_from_downstream( proxy, state );
-#ifdef PREFETCH
 					/* Check if we can fulfil the request from prefetch, or
 					 * rewrite the request to fill the prefetch buffer if needed
 					 */
-					if ( state == WRITE_TO_UPSTREAM ) {
+					if ( proxy_prefetches( proxy ) && state == WRITE_TO_UPSTREAM ) {
 						state = proxy_prefetch_for_request( proxy, state );
 					}
-#endif
 				}
 				break;
 			case CONNECT_TO_UPSTREAM:
@@ -774,12 +814,10 @@ void proxy_session( struct proxier* proxy )
 				if ( FD_ISSET( proxy->upstream_fd, &rfds ) ) {
 					state = proxy_read_from_upstream( proxy, state );
 				}
-# ifdef PREFETCH
 				/* Fill the prefetch buffer and rewrite the reply, if needed */
-				if ( state == WRITE_TO_DOWNSTREAM ) {
+				if ( proxy_prefetches( proxy ) && state == WRITE_TO_DOWNSTREAM ) {
 					state = proxy_prefetch_for_reply( proxy, state );
 				}
-#endif
 				break;
 			case WRITE_TO_DOWNSTREAM:
 				if ( FD_ISSET( proxy->downstream_fd, &wfds ) ) {

src/proxy/proxy.h

@@ -5,7 +5,6 @@
 #include <unistd.h>
 
 #include "ioutil.h"
-#include "flexnbd.h"
 #include "parse.h"
 #include "nbdtypes.h"
 #include "self_pipe.h"
@@ -21,9 +20,6 @@
 #define UPSTREAM_TIMEOUT 30 * 1000
 
 struct proxier {
-	/* The flexnbd wrapper this proxier is attached to */
-	struct flexnbd*   flexnbd;
-
 	/** address/port to bind to */
 	union mysockaddr  listen_on;
 
@@ -48,7 +44,7 @@ struct proxier {
 	int               upstream_fd;
 
 	/* This is the size we advertise to the downstream server */
-	off64_t           upstream_size;
+	uint64_t          upstream_size;
 
 	/* We transform the raw request header into here */
 	struct nbd_request req_hdr;
@@ -73,7 +69,8 @@ struct proxier {
 	uint64_t req_count;
 	int hello_sent;
 
-#ifdef PREFETCH
+	/** These are only used if we pass --cache on the command line */
+
 	/* While the in-flight request has been munged by prefetch, these two are
 	 * set to true, and the original length of the request, respectively */
 	int is_prefetch_req;
@@ -81,7 +78,8 @@ struct proxier {
 
 	/* And here, we actually store the prefetched data once it's returned */
 	struct prefetch *prefetch;
-#endif
+
+	/** */
 };
 
 struct proxier* proxy_create(
@@ -89,7 +87,8 @@ struct proxier* proxy_create(
 	char* s_downstream_port,
 	char* s_upstream_address,
 	char* s_upstream_port,
-	char* s_upstream_bind );
+	char* s_upstream_bind,
+	char* s_cache_bytes);
 int do_proxy( struct proxier* proxy );
 void proxy_cleanup( struct proxier* proxy );
 void proxy_destroy( struct proxier* proxy );

src/server/bitset.h

@@ -7,43 +7,64 @@
 #include <string.h>
 #include <pthread.h>
 
+/*
+ * Make the bitfield words 'opaque' to prevent code
+ * poking at the bits directly without using these
+ * accessors/macros
+ */
+typedef uint64_t	bitfield_word_t;
+typedef bitfield_word_t * bitfield_p;
 
-static inline char char_with_bit_set(uint64_t num) { return 1<<(num%8); }
+#define BITFIELD_WORD_SIZE	sizeof(bitfield_word_t)
+#define BITS_PER_WORD	(BITFIELD_WORD_SIZE * 8)
+
+#define BIT_MASK(_idx)	\
+			(1LL << ((_idx) & (BITS_PER_WORD - 1)))
+#define BIT_WORD(_b, _idx)	\
+			((bitfield_word_t*)(_b))[(_idx) / BITS_PER_WORD]
+
+/* Calculates the number of words needed to store _bytes number of bytes
+ * this is added to accommodate code that wants to use bytes sizes
+ */
+#define BIT_WORDS_FOR_SIZE(_bytes) \
+			((_bytes + (BITFIELD_WORD_SIZE-1)) / BITFIELD_WORD_SIZE)
+
+/** Return the bit value ''idx'' in array ''b'' */
+static inline int bit_get(bitfield_p b, uint64_t idx) {
+	return (BIT_WORD(b, idx) >> (idx & (BITS_PER_WORD-1))) & 1;
+}
 
 /** Return 1 if the bit at ''idx'' in array ''b'' is set */
-static inline int bit_is_set(char* b, uint64_t idx) {
+static inline int bit_is_set(bitfield_p b, uint64_t idx) {
-	return (b[idx/8] & char_with_bit_set(idx)) != 0;
+	return bit_get(b, idx);
 }
 /** Return 1 if the bit at ''idx'' in array ''b'' is clear */
-static inline int bit_is_clear(char* b, uint64_t idx) {
+static inline int bit_is_clear(bitfield_p b, uint64_t idx) {
-	return !bit_is_set(b, idx);
+	return !bit_get(b, idx);
 }
 /** Tests whether the bit at ''idx'' in array ''b'' has value ''value'' */
-static inline int bit_has_value(char* b, uint64_t idx, int value) {
+static inline int bit_has_value(bitfield_p b, uint64_t idx, int value) {
-	if (value) { return bit_is_set(b, idx); }
+	return bit_get(b, idx) == !!value;
-	else { return bit_is_clear(b, idx); }
 }
 /** Sets the bit ''idx'' in array ''b'' */
-static inline void bit_set(char* b, uint64_t idx) {
+static inline void bit_set(bitfield_p b, uint64_t idx) {
-	b[idx/8] |= char_with_bit_set(idx);
+	BIT_WORD(b, idx) |= BIT_MASK(idx);
-	//__sync_fetch_and_or(b+(idx/8), char_with_bit_set(idx));
 }
 /** Clears the bit ''idx'' in array ''b'' */
-static inline void bit_clear(char* b, uint64_t idx) {
+static inline void bit_clear(bitfield_p b, uint64_t idx) {
-	b[idx/8] &= ~char_with_bit_set(idx);
+	BIT_WORD(b, idx) &= ~BIT_MASK(idx);
-	//__sync_fetch_and_nand(b+(idx/8), char_with_bit_set(idx));
 }
 /** Sets ''len'' bits in array ''b'' starting at offset ''from'' */
-static inline void bit_set_range(char* b, uint64_t from, uint64_t len)
+static inline void bit_set_range(bitfield_p b, uint64_t from, uint64_t len)
 {
-	for ( ; from%8 != 0 && len > 0 ; len-- ) {
+	for ( ; (from % BITS_PER_WORD) != 0 && len > 0 ; len-- ) {
 		bit_set( b, from++ );
 	}
 
-	if (len >= 8) {
+	if (len >= BITS_PER_WORD) {
-		memset(b+(from/8), 255, len/8 );
+		memset(&BIT_WORD(b, from), 0xff, len / 8 );
 		from += len;
-		len = (len%8);
+		len = len % BITS_PER_WORD;
 		from -= len;
 	}
 
@@ -52,16 +73,16 @@ static inline void bit_set_range(char* b, uint64_t from, uint64_t len)
 	}
 }
 /** Clears ''len'' bits in array ''b'' starting at offset ''from'' */
-static inline void bit_clear_range(char* b, uint64_t from, uint64_t len)
+static inline void bit_clear_range(bitfield_p b, uint64_t from, uint64_t len)
 {
-	for ( ; from%8 != 0 && len > 0 ; len-- ) {
+	for ( ; (from % BITS_PER_WORD) != 0 && len > 0 ; len-- ) {
 		bit_clear( b, from++ );
 	}
 
-	if (len >= 8) {
+	if (len >= BITS_PER_WORD) {
-		memset(b+(from/8), 0, len/8 );
+		memset(&BIT_WORD(b, from), 0, len / 8 );
 		from += len;
-		len = (len%8);
+		len = len % BITS_PER_WORD;
 		from -= len;
 	}
 
@@ -75,16 +96,16 @@ static inline void bit_clear_range(char* b, uint64_t from, uint64_t len)
   * bits that are the same as the first one specified. If ''run_is_set'' is
   * non-NULL, the value of that bit is placed into it.
   */
-static inline uint64_t bit_run_count(char* b, uint64_t from, uint64_t len, int *run_is_set) {
+static inline uint64_t bit_run_count(bitfield_p b, uint64_t from, uint64_t len, int *run_is_set) {
-	uint64_t* current_block;
 	uint64_t count = 0;
-	int first_value = bit_is_set(b, from);
+	int first_value = bit_get(b, from);
+	bitfield_word_t word_match = first_value ? -1 : 0;
 
 	if ( run_is_set != NULL ) {
 		*run_is_set = first_value;
 	}
 
-	for ( ; (from+count) % 64 != 0 && len > 0; len--) {
+	for ( ; ((from + count) % BITS_PER_WORD) != 0 && len > 0; len--) {
 		if (bit_has_value(b, from + count, first_value)) {
 			count++;
 		} else {
@@ -92,10 +113,9 @@ static inline uint64_t bit_run_count(char* b, uint64_t from, uint64_t len, int *
 		}
 	}
 
-	for ( ; len >= 64 ; len -= 64 ) {
+	for ( ; len >= BITS_PER_WORD ; len -= BITS_PER_WORD ) {
-		current_block = (uint64_t*) (b + ((from+count)/8));
+		if (BIT_WORD(b, from + count) == word_match) {
-		if (*current_block == ( first_value ? UINT64_MAX : 0 ) ) {
+			count += BITS_PER_WORD;
-			count += 64;
 		} else {
 			break;
 		}
@@ -154,7 +174,7 @@ struct bitset {
 	int resolution;
 	struct bitset_stream *stream;
 	int stream_enabled;
-	char bits[];
+	bitfield_word_t bits[];
 };
 
 /** Allocate a bitset for a file of the given size, and chunks of the
@@ -162,9 +182,12 @@ struct bitset {
   */
 static inline struct bitset *bitset_alloc( uint64_t size, int resolution )
 {
-	struct bitset *bitset = xmalloc(
+	// calculate a size to allocate that is a multiple of the size of the
-		sizeof( struct bitset ) + ( size + resolution - 1 ) / resolution
+	// bitfield word
-	);
+	size_t bitfield_size =
+			BIT_WORDS_FOR_SIZE((( size + resolution - 1 ) / resolution)) * sizeof( bitfield_word_t );
+	struct bitset *bitset = xmalloc(sizeof( struct bitset ) + ( bitfield_size / 8 ) );
+
 	bitset->size = size;
 	bitset->resolution = resolution;
 	/* don't actually need to call pthread_mutex_destroy '*/

src/server/client.c

@@ -118,6 +118,7 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 		 */
 
 		uint64_t run = bitset_run_count(map, from, len);
+		struct iommap *iommap = iommap_alloc(client->fileno, from, len); 
 
 		debug("write_not_zeroes: from=%ld, len=%d, run=%d", from, len, run);
 
@@ -126,7 +127,9 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 			debug("(run adjusted to %d)", run);
 		}
 
-		if (0) /* useful but expensive */
+		/* 
+		// Useful but expensive
+		if (0) 
 		{
 			uint64_t i;
 			fprintf(stderr, "full map resolution=%d: ", map->resolution);
@@ -139,6 +142,7 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 			}
 			fprintf(stderr, "\n");
 		}
+		*/
 
 		#define DO_READ(dst, len) ERROR_IF_NEGATIVE( \
 			readloop( \
@@ -152,7 +156,7 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 		if (bitset_is_set_at(map, from)) {
 			debug("writing the lot: from=%ld, run=%d", from, run);
 			/* already allocated, just write it all */
-			DO_READ(client->mapped + from, run);
+			DO_READ(iommap->buf, run);
 			/* We know from our earlier call to  bitset_run_count that the
 			 * bitset is all-1s at this point, but we need to dirty it for the
 			 * sake of the event stream - the actual bytes have changed, and we
@@ -183,7 +187,7 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 					(0 == memcmp( zerobuffer, zerobuffer+1, blockrun-1 ));
 
 				if ( !all_zeros ) {
-					memcpy(client->mapped+from, zerobuffer, blockrun);
+					memcpy(iommap->buf, zerobuffer, blockrun);
 					bitset_set_range(map, from, blockrun);
 					/* at this point we could choose to
 					 * short-cut the rest of the write for
@@ -202,6 +206,8 @@ void write_not_zeroes(struct client* client, uint64_t from, uint64_t len)
 				from += blockrun;
 			}
 		}
+		iommap_sync(iommap);
+		iommap_free(iommap);
 	}
 }
 
@@ -231,6 +237,10 @@ int client_read_request( struct client * client , struct nbd_request *out_reques
 				debug( "Connection reset while"
 						" reading request" );
 				return 0;
+			case ETIMEDOUT:
+				debug( "Connection timed out while"
+						" reading request" );
+				return 0;
 			default:
 				/* FIXME: I've seen this happen, but I
 				 * couldn't reproduce it so I'm leaving
@@ -249,14 +259,14 @@ int client_read_request( struct client * client , struct nbd_request *out_reques
 	return 1;
 }
 
-int fd_write_reply( int fd, char *handle, int error )
+int fd_write_reply( int fd, uint64_t handle, int error )
 {
 	struct nbd_reply     reply;
 	struct nbd_reply_raw reply_raw;
 
 	reply.magic = REPLY_MAGIC;
 	reply.error = error;
-	memcpy( reply.handle, handle, 8 );
+	reply.handle.w = handle;
 
 	nbd_h2r_reply( &reply, &reply_raw );
 	debug( "Replying with handle=0x%08X, error=%"PRIu32, handle, error );
@@ -288,7 +298,7 @@ int fd_write_reply( int fd, char *handle, int error )
  */
 int client_write_reply( struct client * client, struct nbd_request *request, int error )
 {
-	return fd_write_reply( client->socket, request->handle, error);
+	return fd_write_reply( client->socket, request->handle.w, error);
 }
 
 
@@ -297,7 +307,7 @@ void client_write_init( struct client * client, uint64_t size )
 	struct nbd_init init = {{0}};
 	struct nbd_init_raw init_raw = {{0}};
 
-	memcpy( init.passwd, INIT_PASSWD, sizeof( INIT_PASSWD ) );
+	memcpy( init.passwd, INIT_PASSWD, sizeof( init.passwd ) );
 	init.magic = INIT_MAGIC;
 	init.size = size;
 	memset( init.reserved, 0, 128 );
@@ -416,8 +426,8 @@ void client_reply_to_read( struct client* client, struct nbd_request request )
 {
 	off64_t offset;
 
-	// TODO: cork
 	debug("request read %ld+%d", request.from, request.len);
+	sock_set_tcp_cork( client->socket, 1 );
 	client_write_reply( client, &request, 0 );
 
 	offset = request.from;
@@ -435,29 +445,35 @@ void client_reply_to_read( struct client* client, struct nbd_request request )
 			offset,
 			request.len);
 
-	// TODO: uncork
+	sock_set_tcp_cork( client->socket, 0 );
 }
 
 
 void client_reply_to_write( struct client* client, struct nbd_request request )
 {
 	debug("request write from=%"PRIu64", len=%"PRIu32", handle=0x%08X", request.from, request.len, request.handle);
-	if (client->serve->allocation_map_built) {
+	// TODO: Just write directly for now.  Not (yet) convinced my changes later on work.
+	// if (client->serve->allocation_map_built) {
+	if (0) {
 		write_not_zeroes( client, request.from, request.len );
 	}
 	else {
 		debug("No allocation map, writing directly.");
 		/* If we get cut off partway through reading this data:
 		 * */
+		struct iommap *iommap = iommap_alloc(client->fileno, request.from, request.len); 
+
 		ERROR_IF_NEGATIVE(
 			readloop( client->socket,
-				client->mapped + request.from,
+				iommap->buf,
 				request.len),
 			"reading write data failed from=%ld, len=%d",
 			request.from,
 			request.len
 		);
 
+		iommap_sync(iommap);
+		iommap_free(iommap);
 		/* the allocation_map is shared between client threads, and may be
 		 * being built. We need to reflect the write in it, as it may be in
 		 * a position the builder has already gone over.
@@ -465,19 +481,6 @@ void client_reply_to_write( struct client* client, struct nbd_request request )
 		bitset_set_range(client->serve->allocation_map, request.from, request.len);
 	}
 
-	if (1) /* not sure whether this is necessary... */
-	{
-		/* multiple of 4K page size */
-		uint64_t from_rounded = request.from & (!0xfff);
-		uint64_t len_rounded = request.len + (request.from - from_rounded);
-
-		FATAL_IF_NEGATIVE(
-			msync( client->mapped + from_rounded,
-				len_rounded,
-				MS_SYNC | MS_INVALIDATE),
-			"msync failed %ld %ld", request.from, request.len
-		);
-	}
 	client_write_reply( client, &request, 0);
 }
 
@@ -643,9 +646,6 @@ void client_cleanup(struct client* client,
 		debug("Closed client socket fd %d", client->socket);
 		client->socket = -1;
 	}
-	if (client->mapped) {
-		munmap(client->mapped, client->serve->size);
-	}
 	if (client->fileno) {
 		FATAL_IF_NEGATIVE( close(client->fileno),
 			"Error closing file %d",
@@ -661,25 +661,20 @@ void client_cleanup(struct client* client,
 void* client_serve(void* client_uncast)
 {
 	struct client* client = (struct client*) client_uncast;
+	void** a = NULL;
 
 	error_set_handler((cleanup_handler*) client_cleanup, client);
-
 	info("client: mmaping file");
 	FATAL_IF_NEGATIVE(
 		open_and_mmap(
 			client->serve->filename,
 			&client->fileno,
 			NULL,
-			(void**) &client->mapped
+			a
 		),
 		"Couldn't open/mmap file %s: %s", client->serve->filename, strerror( errno )
 	);
 
-	FATAL_IF_NEGATIVE(
-		madvise( client->mapped, client->serve->size, MADV_RANDOM ),
-		SHOW_ERRNO( "Failed to madvise() %s", client->serve->filename )
-	);
-
 	debug( "Opened client file fd %d", client->fileno);
 	debug("client: sending hello");
 	client_send_hello(client);

src/server/client.h

@@ -29,7 +29,6 @@ struct client {
 	int     socket;
 
 	int     fileno;
-	char*   mapped;
 
 	struct self_pipe * stop_signal;
 

src/server/mirror.c

@@ -139,7 +139,7 @@ enum mirror_state mirror_get_state( struct mirror * mirror )
 void mirror_init( struct mirror * mirror, const char * filename )
 {
 	int map_fd;
-	off64_t size;
+	uint64_t size;
 
 	NULLCHECK( mirror );
 	NULLCHECK( filename );
@@ -270,7 +270,7 @@ void mirror_cleanup( struct server * serve,
 }
 
 
-int mirror_connect( struct mirror * mirror, off64_t local_size )
+int mirror_connect( struct mirror * mirror, uint64_t local_size )
 {
 	struct sockaddr * connect_from = NULL;
 	int connected =  0;
@@ -292,7 +292,7 @@ int mirror_connect( struct mirror * mirror, off64_t local_size )
 				"Select failed." );
 
 		if( FD_ISSET( mirror->client, &fds ) ){
-			off64_t remote_size;
+			uint64_t remote_size;
 			if ( socket_nbd_read_hello( mirror->client, &remote_size ) ) {
 				if( remote_size == local_size ){
 					connected = 1;
@@ -412,7 +412,7 @@ int mirror_setup_next_xfer( struct mirror_ctrl *ctrl )
 	struct nbd_request req = {
 		.magic = REQUEST_MAGIC,
 		.type = REQUEST_WRITE,
-		.handle = ".MIRROR.",
+		.handle.b = ".MIRROR.",
 		.from = current,
 		.len = run
 	};
@@ -462,6 +462,12 @@ static void mirror_write_cb( struct ev_loop *loop, ev_io *w, int revents )
 
 	debug( "Mirror write callback invoked with events %d. fd: %i", revents, ctrl->mirror->client );
 
+	/* FIXME: We can end up corking multiple times in unusual circumstances; this
+	 * is annoying, but harmless */
+	if ( xfer->written == 0 ) {
+		sock_set_tcp_cork( ctrl->mirror->client, 1 );
+	}
+
 	if ( xfer->written < hdr_size ) {
 		data_loc = ( (char*) &xfer->hdr.req_raw ) + ctrl->xfer.written;
 		to_write = hdr_size - xfer->written;
@@ -489,6 +495,7 @@ static void mirror_write_cb( struct ev_loop *loop, ev_io *w, int revents )
 
 	// All bytes written, so now we need to read the NBD reply back.
 	if ( ctrl->xfer.written == ctrl->xfer.len + hdr_size ) {
+	  sock_set_tcp_cork( ctrl->mirror->client, 0 ) ;
 		ev_io_start( loop, &ctrl->read_watcher  );
 		ev_io_stop(  loop, &ctrl->write_watcher );
 	}
@@ -561,7 +568,7 @@ static void mirror_read_cb( struct ev_loop *loop, ev_io *w, int revents )
 		return;
 	}
 
-	if ( memcmp( ".MIRROR.", &rsp.handle[0], 8 ) != 0 ) {
+	if ( memcmp( ".MIRROR.", rsp.handle.b, 8 ) != 0 ) {
 		warn( "Bad handle returned from listener" );
 		ev_break( loop, EVBREAK_ONE );
 		return;
@@ -636,7 +643,7 @@ static void mirror_read_cb( struct ev_loop *loop, ev_io *w, int revents )
 	return;
 }
 
-void mirror_timeout_cb( struct ev_loop *loop, ev_timer *w __attribute__((unused)), int revents )
+static void mirror_timeout_cb( struct ev_loop *loop, ev_timer *w __attribute__((unused)), int revents )
 {
 	if ( !(revents & EV_TIMER ) ) {
 		warn( "Mirror timeout called but no timer event signalled" );
@@ -648,7 +655,7 @@ void mirror_timeout_cb( struct ev_loop *loop, ev_timer *w __attribute__((unused)
 	return;
 }
 
-void mirror_abandon_cb( struct ev_loop *loop, ev_io *w, int revents )
+static void mirror_abandon_cb( struct ev_loop *loop, ev_io *w, int revents )
 {
 	struct mirror_ctrl* ctrl = (struct mirror_ctrl*) w->data;
 	NULLCHECK( ctrl );
@@ -666,7 +673,7 @@ void mirror_abandon_cb( struct ev_loop *loop, ev_io *w, int revents )
 }
 
 
-void mirror_limit_cb( struct ev_loop *loop, ev_timer *w, int revents )
+static void mirror_limit_cb( struct ev_loop *loop, ev_timer *w, int revents )
 {
 	struct mirror_ctrl* ctrl = (struct mirror_ctrl*) w->data;
 	NULLCHECK( ctrl );
@@ -694,7 +701,7 @@ void mirror_limit_cb( struct ev_loop *loop, ev_timer *w, int revents )
  * if it has, start migrating. If it's not finished, then enabling the bitset
  * stream does not go well for us.
  */
-void mirror_begin_cb( struct ev_loop *loop, ev_timer *w, int revents )
+static void mirror_begin_cb( struct ev_loop *loop, ev_timer *w, int revents )
 {
 	struct mirror_ctrl* ctrl = (struct mirror_ctrl*) w->data;
 	NULLCHECK( ctrl );
@@ -750,7 +757,8 @@ void mirror_run( struct server *serve )
 
 	ctrl.ev_loop = EV_DEFAULT;
 
-	/* gcc warns on -O2. clang is fine. Seems to be the fault of ev.h */
+	/* gcc warns with -Wstrict-aliasing on -O2. clang doesn't
+	 * implement this warning. Seems to be the fault of ev.h */
 	ev_init( &ctrl.begin_watcher, mirror_begin_cb );
 	ctrl.begin_watcher.repeat = 1.0; // We check bps every second. seems sane.
 	ctrl.begin_watcher.data = (void*) &ctrl;
@@ -914,7 +922,7 @@ void* mirror_runner(void* serve_params_uncast)
 	 * for us ). But if we've failed and are going to retry on the next run, we
 	 * must close this socket here to have any chance of it succeeding.
 	 */
-	if ( !mirror->client < 0 ) {
+	if ( !(mirror->client < 0) ) {
 		sock_try_close( mirror->client );
 		mirror->client = -1;
 	}

src/server/mode.c

@@ -220,7 +220,6 @@ void read_serve_param( int c, char **ip_addr, char **ip_port, char **file, char
 		case 'h':
 			fprintf(stdout, "%s\n", serve_help_text );
 			exit( 0 );
-			break;
 		case 'l':
 			*ip_addr = optarg;
 			break;
@@ -263,7 +262,6 @@ void read_listen_param( int c,
 		case 'h':
 			fprintf(stdout, "%s\n", listen_help_text );
 			exit(0);
-			break;
 		case 'l':
 			*ip_addr = optarg;
 			break;
@@ -297,7 +295,6 @@ void read_readwrite_param( int c, char **ip_addr, char **ip_port, char **bind_ad
 		case 'h':
 			fprintf(stdout, "%s\n", err_text );
 			exit( 0 );
-			break;
 		case 'l':
 			*ip_addr = optarg;
 			break;
@@ -331,7 +328,6 @@ void read_sock_param( int c, char **sock, char *help_text )
 		case 'h':
 			fprintf( stdout, "%s\n", help_text );
 			exit( 0 );
-			break;
 		case 's':
 			*sock = optarg;
 			break;
@@ -362,7 +358,6 @@ void read_mirror_speed_param(
 		case 'h':
 			fprintf( stdout, "%s\n", mirror_speed_help_text );
 			exit( 0 );
-			break;
 		case 's':
 			*sock = optarg;
 			break;
@@ -394,7 +389,6 @@ void read_mirror_param(
 		case 'h':
 			fprintf( stdout, "%s\n", mirror_help_text );
 			exit( 0 );
-			break;
 		case 's':
 			*sock = optarg;
 			break;
@@ -428,7 +422,6 @@ void read_break_param( int c, char **sock )
 		case 'h':
 			fprintf( stdout, "%s\n", break_help_text );
 			exit( 0 );
-			break;
 		case 's':
 			*sock = optarg;
 			break;
@@ -580,7 +573,10 @@ void params_readwrite(
 
 	parse_port( s_port, &out->connect_to.v4 );
 
-	out->from  = atol(s_from);
+	long signed_from = atol(s_from);
+	FATAL_IF_NEGATIVE( signed_from,
+			"Can't read from a negative offset %d.", signed_from);
+	out->from  = signed_from;
 
 	if (write_not_read) {
 		if (s_length_or_filename[0]-48 < 10) {
@@ -592,9 +588,10 @@ void params_readwrite(
 			  s_length_or_filename, O_RDONLY);
 			FATAL_IF_NEGATIVE(out->data_fd,
 			  "Couldn't open %s", s_length_or_filename);
-			out->len = lseek64(out->data_fd, 0, SEEK_END);
+			off64_t signed_len = lseek64(out->data_fd, 0, SEEK_END);
-			FATAL_IF_NEGATIVE(out->len,
+			FATAL_IF_NEGATIVE(signed_len,
 			  "Couldn't find length of %s", s_length_or_filename);
+			out->len = signed_len;
 			FATAL_IF_NEGATIVE(
 				lseek64(out->data_fd, 0, SEEK_SET),
 				"Couldn't rewind %s", s_length_or_filename

src/server/serve.c

@@ -78,6 +78,8 @@ struct server * server_create (
 	NULLCHECK( out->close_signal );
 	NULLCHECK( out->acl_updated_signal );
 
+	log_context = s_file;
+
 	return out;
 }
 
@@ -233,7 +235,6 @@ int tryjoin_client_thread( struct client_tbl_entry *entry, int (*joinfunc)(pthre
 
 	int was_closed = 0;
 	void * status=NULL;
-	int join_errno;
 
 	if (entry->thread != 0) {
 		char s_client_address[128];
@@ -241,7 +242,7 @@ int tryjoin_client_thread( struct client_tbl_entry *entry, int (*joinfunc)(pthre
 		sockaddr_address_string( &entry->address.generic, &s_client_address[0], 128 );
 
 		debug( "%s(%p,...)", joinfunc == pthread_join ? "joining" : "tryjoining", entry->thread );
-		join_errno = joinfunc(entry->thread, &status);
+		int join_errno = joinfunc(entry->thread, &status);
 
 		/* join_errno can legitimately be ESRCH if the thread is
 		 * already dead, but the client still needs tidying up. */
@@ -256,7 +257,7 @@ int tryjoin_client_thread( struct client_tbl_entry *entry, int (*joinfunc)(pthre
 			debug("nbd thread %016x exited (%s) with status %ld",
 					entry->thread,
 					s_client_address,
-					(uint64_t)status);
+					(uintptr_t)status);
 			client_destroy( entry->client );
 			entry->client = NULL;
 			entry->thread = 0;
@@ -423,6 +424,9 @@ void accept_nbd_client(
 	int slot;
 	char s_client_address[64] = {0};
 
+	FATAL_IF_NEGATIVE( sock_set_keepalive_params( client_fd, CLIENT_KEEPALIVE_TIME, CLIENT_KEEPALIVE_INTVL, CLIENT_KEEPALIVE_PROBES),
+		"Error setting keepalive parameters on client socket fd %d", client_fd );
+
 
 	if ( !server_should_accept_client( params, client_address, s_client_address, 64 ) ) {
 		FATAL_IF_NEGATIVE( close( client_fd ),
@@ -598,7 +602,6 @@ int server_accept( struct server * params )
 {
 	NULLCHECK( params );
 	debug("accept loop starting");
-	int              client_fd;
 	union mysockaddr client_address;
 	fd_set           fds;
 	socklen_t        socklen=sizeof(client_address);
@@ -638,7 +641,7 @@ int server_accept( struct server * params )
 	}
 
 	if ( FD_ISSET( params->server_fd, &fds ) ){
-		client_fd = accept( params->server_fd, &client_address.generic, &socklen );
+		int client_fd = accept( params->server_fd, &client_address.generic, &socklen );
 
 		if ( params->allow_new_clients ) {
 			debug("Accepted nbd client socket fd %d", client_fd);
@@ -741,11 +744,11 @@ void server_join_clients( struct server * serve ) {
 
 	for (i=0; i < serve->max_nbd_clients; i++) {
 		pthread_t thread_id = serve->nbd_client[i].thread;
-		int err = 0;
 
 		if (thread_id != 0) {
 			debug( "joining thread %p", thread_id );
-			if ( 0 == (err = pthread_join( thread_id, &status ) ) ) {
+			int err = pthread_join( thread_id, &status );
+			if ( 0 == err ) {
 				serve->nbd_client[i].thread = 0;
 			} else {
 				warn( "Error %s (%i) joining thread %p", strerror( err ), err, thread_id );

src/server/serve.h

@@ -21,6 +21,9 @@ struct client_tbl_entry {
 
 
 #define MAX_NBD_CLIENTS 16
+#define CLIENT_KEEPALIVE_TIME 30
+#define CLIENT_KEEPALIVE_INTVL 10
+#define CLIENT_KEEPALIVE_PROBES 3
 struct server {
 	/* The flexnbd wrapper this server is attached to */
 	struct flexnbd * flexnbd;
@@ -154,8 +157,10 @@ int do_serve( struct server *, struct self_pipe * );
 struct mode_readwrite_params {
 	union mysockaddr     connect_to;
 	union mysockaddr     connect_from;
-	off64_t              from;
+
-	off64_t              len;
+	uint64_t             from;
+	uint32_t             len;
+
 	int                  data_fd;
 	int                  client;
 };

src/server/status.c

@@ -31,6 +31,7 @@ struct status * status_create( struct server * serve )
 		status->migration_speed_limit = serve->mirror->max_bytes_per_second;
 
 		status->migration_seconds_left = server_mirror_eta( serve );
+		status->migration_bytes_left = server_mirror_bytes_remaining( serve );
 	}
 
 	server_unlock_start_mirror( serve );
@@ -60,6 +61,7 @@ int status_write( struct status * status, int fd )
 		PRINT_UINT64( migration_speed );
 		PRINT_UINT64( migration_duration );
 		PRINT_UINT64( migration_seconds_left );
+		PRINT_UINT64( migration_bytes_left );
 		if ( status->migration_speed_limit < UINT64_MAX ) {
 			PRINT_UINT64( migration_speed_limit );
 		};

src/server/status.h

@@ -64,6 +64,8 @@
  *   Our current best estimate of how many seconds are left before the migration
  *   migration is finished.
  *
+ * migration_bytes_left:
+ *   The number of bytes remaining to migrate.
  */
 
 
@@ -84,6 +86,7 @@ struct status {
 	uint64_t migration_speed;
 	uint64_t migration_speed_limit;
 	uint64_t migration_seconds_left;
+	uint64_t migration_bytes_left;
 };
 
 /** Create a status object for the given server. */

tests/acceptance/environment.rb

@@ -21,6 +21,11 @@ class Environment
     @fake_pid = nil
   end
 
+  def prefetch_proxy!
+    @nbd1.prefetch_proxy = true
+    @nbd2.prefetch_proxy = true
+  end
+
   def proxy1(port=@port2)
     @nbd1.proxy(@ip, port)
   end

tests/acceptance/flexnbd.rb

@@ -198,6 +198,8 @@ module FlexNBD
       end
     end
 
+    attr_accessor :prefetch_proxy
+
     def initialize( bin, ip, port )
       @bin  = bin
       @do_debug = ENV['DEBUG']
@@ -208,6 +210,7 @@ module FlexNBD
       @ip = ip
       @port = port
       @kill = []
+      @prefetch_proxy = false
     end
 
 
@@ -247,6 +250,7 @@ module FlexNBD
         "--port #{port} "\
         "--conn-addr #{connect_ip} "\
         "--conn-port #{connect_port} "\
+        "#{prefetch_proxy ? "--cache " : ""}"\
         "#{@debug}"
     end
 

tests/acceptance/flexnbd/constants.rb

@@ -32,7 +32,7 @@ module FlexNBD
       txt_lines.each do |line|
         if line =~ /^#\s*define\s+([A-Z0-9_]+)\s+(\d+)\s*$/
           # Bodge until I can figure out what to do with #ifdefs
-          const_set($1, $2.to_i) unless constants.include?( $1 )
+          const_set($1, $2.to_i) unless const_defined?( $1 )
         end
       end
     end

tests/acceptance/flexnbd/fake_dest.rb

@@ -138,7 +138,7 @@ module FlexNBD
     end
 
 
-    def accept( err_msg = "Timed out waiting for a connection", timeout = 2)
+    def accept( err_msg = "Timed out waiting for a connection", timeout = 5)
       client_sock = nil
 
       begin

tests/acceptance/proxy_tests.rb

@@ -0,0 +1,194 @@
+# encoding: utf-8
+require 'flexnbd/fake_source'
+require 'flexnbd/fake_dest'
+
+module ProxyTests
+  def b
+    "\xFF".b
+  end
+
+  def with_proxied_client( override_size = nil )
+    @env.serve1 unless @server_up
+    @env.proxy2 unless @proxy_up
+    @env.nbd2.can_die(0)
+    client = FlexNBD::FakeSource.new(@env.ip, @env.port2, "Couldn't connect to proxy")
+    begin
+
+      result = client.read_hello
+      assert_equal "NBDMAGIC", result[:magic]
+      assert_equal override_size || @env.file1.size, result[:size]
+
+      yield client
+    ensure
+      client.close rescue nil
+    end
+  end
+
+  def test_exits_with_error_when_cannot_connect_to_upstream_on_start
+    assert_raises(RuntimeError) { @env.proxy1 }
+  end
+
+  def test_read_requests_successfully_proxied
+    with_proxied_client do |client|
+      (0..3).each do |n|
+        offset = n * 4096
+        client.write_read_request(offset, 4096, "myhandle")
+        rsp = client.read_response
+
+        assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
+        assert_equal "myhandle", rsp[:handle]
+        assert_equal 0, rsp[:error]
+
+        orig_data = @env.file1.read(offset, 4096)
+        data = client.read_raw(4096)
+
+        assert_equal 4096, orig_data.size
+        assert_equal 4096, data.size
+
+        assert_equal( orig_data, data,
+                      "Returned data does not match on request #{n+1}" )
+      end
+    end
+  end
+
+  def test_write_requests_successfully_proxied
+    with_proxied_client do |client|
+      (0..3).each do |n|
+        offset = n * 4096
+        client.write(offset, b * 4096)
+        rsp = client.read_response
+
+        assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
+        assert_equal "myhandle", rsp[:handle]
+        assert_equal 0, rsp[:error]
+
+        data = @env.file1.read(offset, 4096)
+
+        assert_equal( ( b * 4096 ), data, "Data not written correctly (offset is #{n})" )
+      end
+    end
+  end
+
+  def make_fake_server
+    server = FlexNBD::FakeDest.new(@env.ip, @env.port1)
+    @server_up = true
+
+    # We return a thread here because accept() and connect() both block for us
+    Thread.new do
+      sc = server.accept # just tell the supervisor we're up
+      sc.write_hello
+
+      [ server, sc ]
+    end
+  end
+
+  def test_read_request_retried_when_upstream_dies_partway
+    maker = make_fake_server
+
+    with_proxied_client(4096) do |client|
+      server, sc1 = maker.value
+
+      # Send the read request to the proxy
+      client.write_read_request( 0, 4096 )
+
+      # ensure we're given the read request
+      req1 = sc1.read_request
+      assert_equal ::FlexNBD::REQUEST_MAGIC, req1[:magic]
+      assert_equal ::FlexNBD::REQUEST_READ, req1[:type]
+      assert_equal 0, req1[:from]
+      assert_not_equal 0, req1[:len]
+
+      # Kill the server again, now we're sure the read request has been sent once
+      sc1.close
+
+      # We expect the proxy to reconnect without our client doing anything.
+      sc2 = server.accept
+      sc2.write_hello
+
+      # And once reconnected, it should resend an identical request.
+      req2 = sc2.read_request
+      assert_equal req1, req2
+
+      # The reply should be proxied back to the client.
+      sc2.write_reply( req2[:handle] )
+      sc2.write_data( b * 4096 )
+
+      # Check it to make sure it's correct
+      rsp = timeout(15) { client.read_response }
+      assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
+      assert_equal 0, rsp[:error]
+      assert_equal req1[:handle], rsp[:handle]
+
+      data = client.read_raw( 4096 )
+      assert_equal( (b * 4096), data, "Wrong data returned" )
+
+      sc2.close
+      server.close
+    end
+
+  end
+
+  def test_write_request_retried_when_upstream_dies_partway
+    maker = make_fake_server
+
+    with_proxied_client(4096) do |client|
+      server, sc1 = maker.value
+
+      # Send the read request to the proxy
+      client.write( 0, ( b * 4096 ) )
+
+      # ensure we're given the read request
+      req1 = sc1.read_request
+      assert_equal ::FlexNBD::REQUEST_MAGIC, req1[:magic]
+      assert_equal ::FlexNBD::REQUEST_WRITE, req1[:type]
+      assert_equal 0, req1[:from]
+      assert_equal 4096, req1[:len]
+      data1 = sc1.read_data( 4096 )
+      assert_equal( ( b * 4096 ), data1, "Data not proxied successfully" )
+
+      # Kill the server again, now we're sure the read request has been sent once
+      sc1.close
+
+      # We expect the proxy to reconnect without our client doing anything.
+      sc2 = server.accept
+      sc2.write_hello
+
+      # And once reconnected, it should resend an identical request.
+      req2 = sc2.read_request
+      assert_equal req1, req2
+      data2 = sc2.read_data( 4096 )
+      assert_equal data1, data2
+
+      # The reply should be proxied back to the client.
+      sc2.write_reply( req2[:handle] )
+
+      # Check it to make sure it's correct
+      rsp = timeout(15) { client.read_response }
+      assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
+      assert_equal 0, rsp[:error]
+      assert_equal req1[:handle], rsp[:handle]
+
+      sc2.close
+      server.close
+    end
+  end
+
+  def test_only_one_client_can_connect_to_proxy_at_a_time
+    with_proxied_client do |client|
+
+      c2 = nil
+      assert_raises(Timeout::Error) do
+        timeout(1) do
+          c2 = FlexNBD::FakeSource.new(@env.ip, @env.port2, "Couldn't connect to proxy (2)")
+          c2.read_hello
+        end
+      end
+      c2.close rescue nil if c2
+    end
+
+
+  end
+
+end
+
+

tests/acceptance/test_happy_path.rb

@@ -115,6 +115,11 @@ class TestHappyPath < Test::Unit::TestCase
 
 
   def test_write_to_high_block
+		# 
+		# This test does not work on 32 bit platforms.
+		#
+		skip("Not relevant on 32-bit platforms") if ( ["a"].pack("p").size < 8 )
+
     # Create a large file, then try to write to somewhere after the 2G boundary
     @env.truncate1 "4G"
     @env.serve1

tests/acceptance/test_prefetch_proxy_mode.rb

@@ -0,0 +1,22 @@
+require 'test/unit'
+require 'environment'
+require 'proxy_tests'
+
+
+class TestPrefetchProxyMode < Test::Unit::TestCase
+  include ProxyTests
+
+  def setup
+    super
+    @env = Environment.new
+    @env.prefetch_proxy!
+    @env.writefile1( "f" * 16 )
+  end
+
+  def teardown
+    @env.cleanup
+    super
+  end
+end
+
+

tests/acceptance/test_proxy_mode.rb

@@ -1,200 +1,20 @@
 require 'test/unit'
 require 'environment'
-require 'flexnbd/fake_source'
+require 'proxy_tests'
-require 'flexnbd/fake_dest'
+
 
 class TestProxyMode < Test::Unit::TestCase
+  include ProxyTests
+
   def setup
     super
     @env = Environment.new
-    @env.writefile1( "0" * 16 )
+    @env.writefile1( "f" * 16 )
   end
 
   def teardown
     @env.cleanup
     super
   end
-
-  def with_proxied_client( override_size = nil )
-    @env.serve1 unless @server_up
-    @env.proxy2 unless @proxy_up
-    @env.nbd2.can_die(0)
-    client = FlexNBD::FakeSource.new(@env.ip, @env.port2, "Couldn't connect to proxy")
-    begin
-
-      result = client.read_hello
-      assert_equal "NBDMAGIC", result[:magic]
-      assert_equal override_size || @env.file1.size, result[:size]
-
-      yield client
-    ensure
-      client.close rescue nil
-    end
-  end
-
-  def test_exits_with_error_when_cannot_connect_to_upstream_on_start
-    assert_raises(RuntimeError) { @env.proxy1 }
-  end
-
-  def test_read_requests_successfully_proxied
-    with_proxied_client do |client|
-      (0..3).each do |n|
-        offset = n * 4096
-        client.write_read_request(offset, 4096, "myhandle")
-        rsp = client.read_response
-
-        assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
-        assert_equal "myhandle", rsp[:handle]
-        assert_equal 0, rsp[:error]
-
-        orig_data = @env.file1.read(offset, 4096)
-        data = client.read_raw(4096)
-
-        assert_equal 4096, orig_data.size
-        assert_equal 4096, data.size
-
-        assert_equal( orig_data, data, "Returned data does not match" )
-      end
-    end
-  end
-
-  def test_write_requests_successfully_proxied
-    with_proxied_client do |client|
-      (0..3).each do |n|
-        offset = n * 4096
-        client.write(offset, "\xFF" * 4096)
-        rsp = client.read_response
-
-        assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
-        assert_equal "myhandle", rsp[:handle]
-        assert_equal 0, rsp[:error]
-
-        data = @env.file1.read(offset, 4096)
-
-        assert_equal( ( "\xFF" * 4096 ), data, "Data not written correctly (offset is #{n})" )
-      end
-    end
-  end
-
-  def make_fake_server
-    server = FlexNBD::FakeDest.new(@env.ip, @env.port1)
-    @server_up = true
-
-    # We return a thread here because accept() and connect() both block for us
-    Thread.new do
-      sc = server.accept # just tell the supervisor we're up
-      sc.write_hello
-
-      [ server, sc ]
-    end
-  end
-
-  def test_read_request_retried_when_upstream_dies_partway
-    maker = make_fake_server
-
-    with_proxied_client(4096) do |client|
-      server, sc1 = maker.value
-
-      # Send the read request to the proxy
-      client.write_read_request( 0, 4096 )
-
-      # ensure we're given the read request
-      req1 = sc1.read_request
-      assert_equal ::FlexNBD::REQUEST_MAGIC, req1[:magic]
-      assert_equal ::FlexNBD::REQUEST_READ, req1[:type]
-      assert_equal 0, req1[:from]
-      assert_not_equal 0, req1[:len]
-
-      # Kill the server again, now we're sure the read request has been sent once
-      sc1.close
-
-      # We expect the proxy to reconnect without our client doing anything.
-      sc2 = server.accept
-      sc2.write_hello
-
-      # And once reconnected, it should resend an identical request.
-      req2 = sc2.read_request
-      assert_equal req1, req2
-
-      # The reply should be proxied back to the client.
-      sc2.write_reply( req2[:handle] )
-      sc2.write_data( "\xFF" * 4096 )
-
-      # Check it to make sure it's correct
-      rsp = timeout(15) { client.read_response }
-      assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
-      assert_equal 0, rsp[:error]
-      assert_equal req1[:handle], rsp[:handle]
-
-      data = client.read_raw( 4096 )
-      assert_equal( ("\xFF" * 4096), data, "Wrong data returned" )
-
-      sc2.close
-      server.close
-    end
-
-  end
-
-  def test_write_request_retried_when_upstream_dies_partway
-    maker = make_fake_server
-
-    with_proxied_client(4096) do |client|
-      server, sc1 = maker.value
-
-      # Send the read request to the proxy
-      client.write( 0, ( "\xFF" * 4096 ) )
-
-      # ensure we're given the read request
-      req1 = sc1.read_request
-      assert_equal ::FlexNBD::REQUEST_MAGIC, req1[:magic]
-      assert_equal ::FlexNBD::REQUEST_WRITE, req1[:type]
-      assert_equal 0, req1[:from]
-      assert_equal 4096, req1[:len]
-      data1 = sc1.read_data( 4096 )
-      assert_equal( ( "\xFF" * 4096 ), data1, "Data not proxied successfully" )
-
-      # Kill the server again, now we're sure the read request has been sent once
-      sc1.close
-
-      # We expect the proxy to reconnect without our client doing anything.
-      sc2 = server.accept
-      sc2.write_hello
-
-      # And once reconnected, it should resend an identical request.
-      req2 = sc2.read_request
-      assert_equal req1, req2
-      data2 = sc2.read_data( 4096 )
-      assert_equal data1, data2
-
-      # The reply should be proxied back to the client.
-      sc2.write_reply( req2[:handle] )
-
-      # Check it to make sure it's correct
-      rsp = timeout(15) { client.read_response }
-      assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
-      assert_equal 0, rsp[:error]
-      assert_equal req1[:handle], rsp[:handle]
-
-      sc2.close
-      server.close
-    end
-  end
-
-  def test_only_one_client_can_connect_to_proxy_at_a_time
-    with_proxied_client do |client|
-
-      c2 = nil
-      assert_raises(Timeout::Error) do
-        timeout(1) do
-          c2 = FlexNBD::FakeSource.new(@env.ip, @env.port2, "Couldn't connect to proxy (2)")
-          c2.read_hello
-        end
-      end
-      c2.close rescue nil if c2
-    end
-
-
-  end
-
 end
 

tests/acceptance/test_serve_mode.rb

@@ -6,6 +6,7 @@ class TestServeMode < Test::Unit::TestCase
 
   def setup
     super
+    @b = "\xFF".b
     @env = Environment.new
     @env.writefile1( "0" )
     @env.serve1
@@ -53,18 +54,18 @@ class TestServeMode < Test::Unit::TestCase
       assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
       assert_equal 0, rsp[:error]
 
-      client.write( 0, "\xFF" )
+      client.write( 0, @b )
       rsp = client.read_response
       assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
       assert_equal 0, rsp[:error]
 
-      client.write( 0, "\xFF\xFF" )
+      client.write( 0, @b * 2 )
       rsp = client.read_response
       assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
       assert_equal 0, rsp[:error]
     end
 
-    assert_equal "\xFF\xFF", @env.file1.read( 0, 2 )
+    assert_equal @b * 2, @env.file1.read( 0, 2 )
   end
 
 

tests/unit/check_bitset.c

@@ -10,7 +10,7 @@
 START_TEST(test_bit_set)
 {
 	uint64_t num = 0;
-	char *bits = (char*) #
+	bitfield_p bits = (bitfield_p) #
 
 #define TEST_BIT_SET(bit, newvalue) \
 	bit_set(bits, (bit)); \
@@ -27,7 +27,7 @@ END_TEST
 START_TEST(test_bit_clear)
 {
 	uint64_t num = 0xffffffffffffffff;
-	char *bits = (char*) #
+	bitfield_p bits = (bitfield_p) #
 
 #define TEST_BIT_CLEAR(bit, newvalue) \
 	bit_clear(bits, (bit)); \
@@ -44,7 +44,7 @@ END_TEST
 START_TEST(test_bit_tests)
 {
 	uint64_t num = 0x5555555555555555;
-	char *bits = (char*) #
+	bitfield_p bits = (bitfield_p) #
 
 	fail_unless(bit_has_value(bits, 0, 1), "bit_has_value malfunction");
 	fail_unless(bit_has_value(bits, 1, 0), "bit_has_value malfunction");
@@ -58,8 +58,8 @@ END_TEST
 
 START_TEST(test_bit_ranges)
 {
-	char buffer[4160];
+	bitfield_word_t buffer[BIT_WORDS_FOR_SIZE(4160)];
-	uint64_t  *longs = (unsigned long*) buffer;
+	uint64_t  *longs = (uint64_t *) buffer;
 	uint64_t i;
 
 	memset(buffer, 0, 4160);
@@ -67,9 +67,9 @@ START_TEST(test_bit_ranges)
 	for (i=0; i<64; i++) {
 		bit_set_range(buffer, i*64, i);
 		fail_unless(
-			longs[i] == (1UL<<i)-1,
+			longs[i] == (1ULL<<i)-1,
 			"longs[%ld] = %lx SHOULD BE %lx",
-			i, longs[i], (1L<<i)-1
+			i, longs[i], (1ULL<<i)-1
 		);
 
 		fail_unless(longs[i+1] == 0, "bit_set_range overshot at i=%d", i);
@@ -84,7 +84,7 @@ END_TEST
 
 START_TEST(test_bit_runs)
 {
-	char buffer[256];
+	bitfield_word_t buffer[BIT_WORDS_FOR_SIZE(256)];
 	int i, ptr=0, runs[] = {
 		56,97,22,12,83,1,45,80,85,51,64,40,63,67,75,64,94,81,79,62
 	};

tests/unit/check_mbox.c

@@ -66,9 +66,9 @@ START_TEST( test_receive_blocks_until_post )
 END_TEST
 
 
-Suite* acl_suite(void)
+Suite* mbox_suite(void)
 {
-	Suite *s = suite_create("acl");
+	Suite *s = suite_create("mbox");
 	TCase *tc_create = tcase_create("create");
 	TCase *tc_post = tcase_create("post");
 
@@ -93,7 +93,7 @@ int main(void)
 	log_level = 2;
 #endif
 	int number_failed;
-	Suite *s = acl_suite();
+	Suite *s = mbox_suite();
 	SRunner *sr = srunner_create(s);
 	srunner_run_all(sr, CK_NORMAL);
 	log_level = 0;

tests/unit/check_nbdtypes.c

@@ -88,14 +88,14 @@ START_TEST(test_request_handle)
 	struct nbd_request_raw request_raw;
 	struct nbd_request     request;
 
-	memcpy( request_raw.handle, "MYHANDLE", 8 );
+	memcpy( request_raw.handle.b, "MYHANDLE", 8 );
 	
 	nbd_r2h_request( &request_raw, &request );
-	memset( request_raw.handle, 0, 8 );
+	request_raw.handle.w = 0;
 	nbd_h2r_request( &request, &request_raw );
   
-	fail_unless( memcmp( request.handle, "MYHANDLE", 8 ) == 0, "The handle was not copied." );
+	fail_unless( memcmp( request.handle.b, "MYHANDLE", 8 ) == 0, "The handle was not copied." );
-	fail_unless( memcmp( request_raw.handle, "MYHANDLE", 8 ) == 0, "The handle was not copied back." );
+	fail_unless( memcmp( request_raw.handle.b, "MYHANDLE", 8 ) == 0, "The handle was not copied back." );
 }
 END_TEST
 
@@ -170,14 +170,14 @@ START_TEST(test_reply_handle)
 	struct nbd_reply_raw reply_raw;
 	struct nbd_reply     reply;
 
-	memcpy( reply_raw.handle, "MYHANDLE", 8 );
+	memcpy( reply_raw.handle.b, "MYHANDLE", 8 );
 	
 	nbd_r2h_reply( &reply_raw, &reply );
-	memset( reply_raw.handle, 0, 8 );
+	reply_raw.handle.w = 0;
 	nbd_h2r_reply( &reply, &reply_raw );
   
-	fail_unless( memcmp( reply.handle, "MYHANDLE", 8 ) == 0, "The handle was not copied." );
+	fail_unless( memcmp( reply.handle.b, "MYHANDLE", 8 ) == 0, "The handle was not copied." );
-	fail_unless( memcmp( reply_raw.handle, "MYHANDLE", 8 ) == 0, "The handle was not copied back." );
+	fail_unless( memcmp( reply_raw.handle.b, "MYHANDLE", 8 ) == 0, "The handle was not copied back." );
 }
 END_TEST
 
@@ -188,14 +188,15 @@ START_TEST( test_convert_from )
 	 * nbd_request_raw */
 	struct nbd_request_raw request_raw;
 	struct nbd_request     request;
-	char readbuf[] = {0x80, 0, 0, 0, 0, 0, 0, 0};
 
-	memcpy( &request_raw.from, readbuf, 8 );
+	uint64_t target = 0x8000000000000000;
 
+	/* this is stored big-endian */
+	request_raw.from = htobe64(target);
+
+	/* We expect this to convert big-endian to the host format */
 	nbd_r2h_request( &request_raw, &request );
 
-	uint64_t target = 1;
-	target <<= 63;
 	fail_unless( target == request.from, "from was wrong" );
 }	
 END_TEST

tests/unit/check_readwrite.c

@@ -22,7 +22,7 @@
 
 
 int fd_read_request( int, struct nbd_request_raw *);
-int fd_write_reply( int, char *, int );
+int fd_write_reply( int, uint64_t, int );
 
 int marker;
 
@@ -46,8 +46,7 @@ void * responder( void *respond_uncast )
 	struct respond * resp = (struct respond *) respond_uncast;
 	int sock_fd = resp->sock_fds[1];
 	struct nbd_request_raw request_raw;
-	char wrong_handle[] = "WHOOPSIE";
+	uint64_t wrong_handle = 0x80;
-
 
 	if( fd_read_request( sock_fd, &request_raw ) == -1){
 		fprintf(stderr, "Problem with fd_read_request\n");
@@ -57,7 +56,7 @@ void * responder( void *respond_uncast )
 			fd_write_reply( sock_fd, wrong_handle, 0 );
 		}
 		else {
-			fd_write_reply( sock_fd, resp->received.handle, 0 );
+			fd_write_reply( sock_fd, resp->received.handle.w, 0 );
 		}
 		write( sock_fd, "12345678", 8 );
 	}
@@ -150,7 +149,7 @@ END_TEST
 
 Suite* readwrite_suite(void)
 {
-	Suite *s = suite_create("acl");
+	Suite *s = suite_create("readwrite");
 	TCase *tc_transfer = tcase_create("entrust");
 	TCase *tc_disconnect = tcase_create("disconnect");
 

tests/unit/check_serve.c

@@ -93,7 +93,7 @@ END_TEST
 
 int connect_client( char *addr, int actual_port, char *source_addr )
 {
-	int client_fd;
+	int client_fd = -1;
 
 	struct addrinfo hint;
 	struct addrinfo *ailist, *aip;

tests/unit/check_status.c

@@ -308,6 +308,7 @@ START_TEST( test_renders_migration_statistics )
 	status.migration_speed = 40000000;
 	status.migration_speed_limit = 40000001;
 	status.migration_seconds_left = 1;
+	status.migration_bytes_left = 5000;
 
 	status_write( &status, fds[1] );
 	fail_if_rendered( fds[0], "migration_duration" );
@@ -335,6 +336,9 @@ START_TEST( test_renders_migration_statistics )
 	status_write( &status, fds[1] );
 	fail_unless_rendered( fds[0], "migration_seconds_left=1" );
 
+	status_write( &status, fds[1] );
+	fail_unless_rendered( fds[0], "migration_bytes_left=5000" );
+
 	status.migration_speed_limit = UINT64_MAX;
 
 	status_write( &status, fds[1] );

tests/unit/check_util.c

@@ -141,9 +141,9 @@ START_TEST( test_fatal_doesnt_call_handler )
 END_TEST
 
 
-Suite* error_suite(void)
+Suite* util_suite(void)
 {
-	Suite *s = suite_create("error");
+	Suite *s = suite_create("util");
 	TCase *tc_process = tcase_create("process");
 	TCase *tc_handler = tcase_create("handler");
 
@@ -163,7 +163,7 @@ Suite* error_suite(void)
 int main(void)
 {
 	int number_failed;
-	Suite *s = error_suite();
+	Suite *s = util_suite();
 	SRunner *sr = srunner_create(s);
 	srunner_run_all(sr, CK_NORMAL);
 	number_failed = srunner_ntests_failed(sr);