bytemark/flexnbd-c
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'try-flags' into 'develop'

Browse Source 
Set flags to show we can accept FUA and FLUSH commands

See merge request open-source/flexnbd-c!38
This commit is contained in:
James Carter
2018-02-08 11:18:31 +00:00
parent 084d429961 f71b872622
commit 806de13024
56 changed files with 989 additions and 940 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Makefile
View File

@@ -109,7 +109,6 @@ install:
clean:
rm -rf build/*
.PHONY: clean objs check_objs all server proxy check_bins check doc build test acceptance
# Include extra dependencies at the end, NOT before 'all'

5
debian/changelog vendored
View File

@@ -1,4 +1,4 @@
flexnbd (0.1.8) UNRELEASED; urgency=medium
flexnbd (0.2.0) UNRELEASED; urgency=medium
[ James Carter ]
* Set TCP keepalive on sockets so broken connections are reaped (#33, !33,
@@ -8,6 +8,9 @@ flexnbd (0.1.8) UNRELEASED; urgency=medium
[ Chris Cottam ]
* Increased NBD_MAX_SIZE from 1MB to 32MB for qemu 2.11 (!35)
[ Patrick J Cherry ]
* Added FLUSH and FUA support (!38)
-- James Carter <james.carter@bytemark.co.uk> Thu, 11 Jan 2018 10:05:35 +0000
flexnbd (0.1.7) stable; urgency=medium

2
src/common/ioutil.c
View File

@@ -85,7 +85,7 @@ int open_and_mmap(const char* filename, int* out_fd, uint64_t *out_size, void **
off64_t size;
/* O_DIRECT should not be used with mmap() */
*out_fd = open(filename, O_RDWR | O_SYNC );
*out_fd = open(filename, O_RDWR | O_NOATIME );
if (*out_fd < 1) {
warn("open(%s) failed: does it exist?", filename);

32
src/common/nbdtypes.c
View File

@@ -13,6 +13,7 @@ void nbd_r2h_init( struct nbd_init_raw * from, struct nbd_init * to )
memcpy( to->passwd, from->passwd, 8 );
to->magic = be64toh( from->magic );
to->size = be64toh( from->size );
to->flags = be32toh( from->flags );
}
void nbd_h2r_init( struct nbd_init * from, struct nbd_init_raw * to)
@@ -20,39 +21,42 @@ void nbd_h2r_init( struct nbd_init * from, struct nbd_init_raw * to)
memcpy( to->passwd, from->passwd, 8 );
to->magic = htobe64( from->magic );
to->size = htobe64( from->size );
to->flags = htobe32( from->flags );
}
void nbd_r2h_request( struct nbd_request_raw *from, struct nbd_request * to )
{
to->magic = htobe32( from->magic );
to->type = htobe32( from->type );
to->handle.w = from->handle.w;
to->from = htobe64( from->from );
to->len = htobe32( from->len );
}
void nbd_h2r_request( struct nbd_request * from, struct nbd_request_raw * to )
{
to->magic = be32toh( from->magic );
to->type = be32toh( from->type );
to->flags = be16toh( from->flags );
to->type = be16toh( from->type );
to->handle.w = from->handle.w;
to->from = be64toh( from->from );
to->len = be32toh( from->len );
}
void nbd_r2h_reply( struct nbd_reply_raw * from, struct nbd_reply * to )
void nbd_h2r_request( struct nbd_request * from, struct nbd_request_raw * to )
{
to->magic = htobe32( from->magic );
to->error = htobe32( from->error );
to->flags = htobe16( from->flags );
to->type = htobe16( from->type );
to->handle.w = from->handle.w;
to->from = htobe64( from->from );
to->len = htobe32( from->len );
}
void nbd_h2r_reply( struct nbd_reply * from, struct nbd_reply_raw * to )
void nbd_r2h_reply( struct nbd_reply_raw * from, struct nbd_reply * to )
{
to->magic = be32toh( from->magic );
to->error = be32toh( from->error );
to->handle.w = from->handle.w;
}
void nbd_h2r_reply( struct nbd_reply * from, struct nbd_reply_raw * to )
{
to->magic = htobe32( from->magic );
to->error = htobe32( from->error );
to->handle.w = from->handle.w;
}

43
src/common/nbdtypes.h
View File

@@ -7,15 +7,38 @@
#define INIT_MAGIC 0x0000420281861253
#define REQUEST_MAGIC 0x25609513
#define REPLY_MAGIC 0x67446698
#define REQUEST_READ 0
#define REQUEST_WRITE 1
#define REQUEST_DISCONNECT 2
#define REQUEST_FLUSH 3
/* The top 2 bytes of the type field are overloaded and can contain flags */
#define REQUEST_MASK 0x0000ffff
/* values for transmission flag field */
#define FLAG_HAS_FLAGS (1 << 0) /* Flags are there */
#define FLAG_SEND_FLUSH (1 << 2) /* Send FLUSH */
#define FLAG_SEND_FUA (1 << 3) /* Send FUA (Force Unit Access) */
/* values for command flag field */
#define CMD_FLAG_FUA (1 << 0)
#if 0
/* Not yet implemented by flexnbd */
#define REQUEST_TRIM 4
#define REQUEST_WRITE_ZEROES 6
#define FLAG_READ_ONLY (1 << 1) /* Device is read-only */
#define FLAG_ROTATIONAL (1 << 4) /* Use elevator algorithm - rotational media */
#define FLAG_SEND_TRIM (1 << 5) /* Send TRIM (discard) */
#define FLAG_SEND_WRITE_ZEROES (1 << 6) /* Send NBD_CMD_WRITE_ZEROES */
#define FLAG_CAN_MULTI_CONN (1 << 8) /* multiple connections are okay */
#define CMD_FLAG_NO_HOLE (1 << 1)
#endif
/* 1MiB is the de-facto standard for maximum size of header + data */
/* 32 MiB is the maximum qemu will send you:
* https://github.com/qemu/qemu/blob/v2.11.0/include/block/nbd.h#L183
*/
#define NBD_MAX_SIZE ( 32 * 1024 * 1024 )
#define NBD_REQUEST_SIZE ( sizeof( struct nbd_request_raw ) )
@@ -38,12 +61,14 @@ struct nbd_init_raw {
char passwd[8];
__be64 magic;
__be64 size;
char reserved[128];
__be32 flags;
char reserved[124];
};
struct nbd_request_raw {
__be32 magic;
__be32 type; /* == READ || == WRITE */
__be16 flags;
__be16 type; /* == READ || == WRITE || == FLUSH */
nbd_handle_t handle;
__be64 from;
__be32 len;
@@ -55,18 +80,18 @@ struct nbd_reply_raw {
nbd_handle_t handle; /* handle you got from request */
};
struct nbd_init {
char passwd[8];
uint64_t magic;
uint64_t size;
char reserved[128];
uint32_t flags;
char reserved[124];
};
struct nbd_request {
uint32_t magic;
uint32_t type; /* == READ || == WRITE || == DISCONNECT */
uint16_t flags;
uint16_t type; /* == READ || == WRITE || == DISCONNECT || == FLUSH */
nbd_handle_t handle;
uint64_t from;
uint32_t len;

31
src/common/readwrite.c
View File

@@ -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, uint64_t* out_size )
int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size, uint32_t* out_flags )
{
if ( strncmp( init_raw->passwd, INIT_PASSWD, 8 ) != 0 ) {
warn( "wrong passwd" );
@@ -56,13 +56,17 @@ int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size )
*out_size = be64toh( init_raw->size );
}
if ( NULL != out_flags ) {
*out_flags = be32toh( init_raw->flags );
}
return 1;
fail:
return 0;
}
int socket_nbd_read_hello( int fd, uint64_t* out_size )
int socket_nbd_read_hello( int fd, uint64_t* out_size, uint32_t* out_flags )
{
struct nbd_init_raw init_raw;
@@ -72,16 +76,17 @@ int socket_nbd_read_hello( int fd, uint64_t* out_size )
return 0;
}
return nbd_check_hello( &init_raw, out_size );
return nbd_check_hello( &init_raw, out_size, out_flags );
}
void nbd_hello_to_buf( struct nbd_init_raw *buf, off64_t out_size )
void nbd_hello_to_buf( struct nbd_init_raw *buf, off64_t out_size, uint32_t out_flags )
{
struct nbd_init init;
memcpy( &init.passwd, INIT_PASSWD, 8 );
init.magic = INIT_MAGIC;
init.size = out_size;
init.flags = out_flags;
memset( buf, 0, sizeof( struct nbd_init_raw ) ); // ensure reserved is 0s
nbd_h2r_init( &init, buf );
@@ -89,10 +94,10 @@ void nbd_hello_to_buf( struct nbd_init_raw *buf, off64_t out_size )
return;
}
int socket_nbd_write_hello(int fd, off64_t out_size)
int socket_nbd_write_hello( int fd, off64_t out_size, uint32_t out_flags )
{
struct nbd_init_raw init_raw;
nbd_hello_to_buf( &init_raw, out_size );
nbd_hello_to_buf( &init_raw, out_size, out_flags );
if ( 0 > writeloop( fd, &init_raw, sizeof( init_raw ) ) ) {
warn( SHOW_ERRNO( "failed to write hello to socket" ) );
@@ -101,10 +106,11 @@ int socket_nbd_write_hello(int fd, off64_t out_size)
return 1;
}
void fill_request(struct nbd_request *request, int type, uint64_t from, uint32_t len)
void fill_request(struct nbd_request *request, uint16_t type, uint16_t flags, uint64_t from, uint32_t len)
{
request->magic = htobe32(REQUEST_MAGIC);
request->type = htobe32(type);
request->type = htobe16(type);
request->flags = htobe16(flags);
request->handle.w = (((uint64_t)rand()) << 32) | ((uint64_t)rand());
request->from = htobe64(from);
request->len = htobe32(len);
@@ -153,7 +159,7 @@ void socket_nbd_read(int fd, uint64_t from, uint32_t len, int out_fd, void* out_
struct nbd_request request;
struct nbd_reply reply;
fill_request(&request, REQUEST_READ, from, len);
fill_request(&request, REQUEST_READ, 0, from, len);
FATAL_IF_NEGATIVE(writeloop(fd, &request, sizeof(request)),
"Couldn't write request");
@@ -177,7 +183,7 @@ void socket_nbd_write(int fd, uint64_t from, uint32_t len, int in_fd, void* in_b
struct nbd_request request;
struct nbd_reply reply;
fill_request(&request, REQUEST_WRITE, from, len);
fill_request(&request, REQUEST_WRITE, 0, from, len);
ERROR_IF_NEGATIVE(writeloop(fd, &request, sizeof(request)),
"Couldn't write request");
@@ -202,7 +208,7 @@ int socket_nbd_disconnect( int fd )
int success = 1;
struct nbd_request request;
fill_request( &request, REQUEST_DISCONNECT, 0, 0 );
fill_request( &request, REQUEST_DISCONNECT, 0, 0, 0 );
/* FIXME: This shouldn't be a FATAL error. We should just drop
* the mirror without affecting the main server.
*/
@@ -213,7 +219,8 @@ int socket_nbd_disconnect( int fd )
#define CHECK_RANGE(error_type) { \
uint64_t size;\
int success = socket_nbd_read_hello(params->client, &size); \
uint32_t flags;\
int success = socket_nbd_read_hello(params->client, &size, &flags); \
if ( success ) {\
uint64_t endpoint = params->from + params->len; \
if (endpoint > size || \

8
src/common/readwrite.h
View File

@@ -7,8 +7,8 @@
#include "nbdtypes.h"
int socket_connect(struct sockaddr* to, struct sockaddr* from);
int socket_nbd_read_hello(int fd, uint64_t* size);
int socket_nbd_write_hello(int fd, uint64_t size);
int socket_nbd_read_hello(int fd, uint64_t* size, uint32_t* flags);
int socket_nbd_write_hello(int fd, uint64_t size, uint32_t flags);
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, uint64_t from, uint32_t len, int out_fd, void* out_buf, int timeout_secs);
int socket_nbd_disconnect( int fd );
@@ -16,8 +16,8 @@ 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, uint64_t out_size );
int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size );
void nbd_hello_to_buf( struct nbd_init_raw* buf, uint64_t out_size, uint32_t out_flags );
int nbd_check_hello( struct nbd_init_raw* init_raw, uint64_t* out_size, uint32_t* out_flags );
#endif

42
src/proxy/proxy.c
View File

@@ -106,7 +106,7 @@ void proxy_destroy( struct proxier* proxy )
}
/* Shared between our two different connect_to_upstream paths */
void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size );
void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size, uint32_t flags );
/* 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.
@@ -120,12 +120,13 @@ int proxy_connect_to_upstream( struct proxier* proxy )
int fd = socket_connect( &proxy->connect_to.generic, connect_from );
uint64_t size = 0;
uint32_t flags = 0;
if ( -1 == fd ) {
return 0;
}
if( !socket_nbd_read_hello( fd, &size ) ) {
if( !socket_nbd_read_hello( fd, &size, &flags ) ) {
WARN_IF_NEGATIVE(
sock_try_close( fd ),
"Couldn't close() after failed read of NBD hello on fd %i", fd
@@ -135,7 +136,7 @@ int proxy_connect_to_upstream( struct proxier* proxy )
proxy->upstream_fd = fd;
sock_set_nonblock( fd, 1 );
proxy_finish_connect_to_upstream( proxy, size );
proxy_finish_connect_to_upstream( proxy, size, flags );
return 1;
}
@@ -191,7 +192,7 @@ error:
return;
}
void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size ) {
void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size, uint32_t flags ) {
if ( proxy->upstream_size == 0 ) {
info( "Size of upstream image is %"PRIu64" bytes", size );
@@ -204,6 +205,17 @@ void proxy_finish_connect_to_upstream( struct proxier *proxy, uint64_t size ) {
proxy->upstream_size = size;
if ( proxy->upstream_flags == 0 ) {
info( "Upstream transmission flags set to %"PRIu32"", flags );
} else if ( proxy->upstream_flags != flags ) {
warn(
"Upstream transmission flags changed from %"PRIu32" to %"PRIu32"",
proxy->upstream_flags, flags
);
}
proxy->upstream_flags = flags;
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" ) );
@@ -305,7 +317,7 @@ int proxy_prefetch_for_request( struct proxier* proxy, int state )
struct nbd_request_raw* req_raw = (struct nbd_request_raw*) proxy->req.buf;
struct nbd_reply_raw *rsp_raw = (struct nbd_reply_raw*) proxy->rsp.buf;
int is_read = ( req->type & REQUEST_MASK ) == REQUEST_READ;
int is_read = req->type == REQUEST_READ;
if ( is_read ) {
/* See if we can respond with what's in our prefetch
@@ -435,19 +447,19 @@ int proxy_read_from_downstream( struct proxier *proxy, int state )
if ( proxy->req.needle == NBD_REQUEST_SIZE ) {
nbd_r2h_request( request_raw, request );
if ( ( request->type & REQUEST_MASK ) == REQUEST_DISCONNECT ) {
if ( request->type == REQUEST_DISCONNECT ) {
info( "Received disconnect request from client" );
return EXIT;
}
/* Simple validations */
if ( ( request->type & REQUEST_MASK ) == REQUEST_READ ) {
if ( request->type == REQUEST_READ ) {
if (request->len > ( NBD_MAX_SIZE - NBD_REPLY_SIZE ) ) {
warn( "NBD read request size %"PRIu32" too large", request->len );
return EXIT;
}
}
if ( (request->type & REQUEST_MASK ) == REQUEST_WRITE ) {
if ( request->type == REQUEST_WRITE ) {
if (request->len > ( NBD_MAX_SIZE - NBD_REQUEST_SIZE ) ) {
warn( "NBD write request size %"PRIu32" too large", request->len );
return EXIT;
@@ -459,8 +471,8 @@ int proxy_read_from_downstream( struct proxier *proxy, int state )
if ( proxy->req.needle == proxy->req.size ) {
debug(
"Received NBD request from downstream. type=%"PRIu32" from=%"PRIu64" len=%"PRIu32,
request->type, request->from, request->len
"Received NBD request from downstream. type=%"PRIu16" flags=%"PRIu16" from=%"PRIu64" len=%"PRIu32,
request->type, request->flags, request->from, request->len
);
/* Finished reading, so advance state. Leave size untouched so the next
@@ -516,11 +528,15 @@ int proxy_read_init_from_upstream( struct proxier* proxy, int state )
if ( proxy->init.needle == proxy->init.size ) {
uint64_t upstream_size;
if ( !nbd_check_hello( (struct nbd_init_raw*) proxy->init.buf, &upstream_size ) ) {
uint32_t upstream_flags;
if ( !nbd_check_hello( (struct nbd_init_raw*) proxy->init.buf, &upstream_size, &upstream_flags ) ) {
warn( "Upstream sent invalid init" );
goto disconnect;
}
/* record the flags, and log the reconnection, set TCP_NODELAY */
proxy_finish_connect_to_upstream( proxy, upstream_size, upstream_flags );
/* Currently, we only get disconnected from upstream (so needing to come
* here) when we have an outstanding request. If that becomes false,
* we'll need to choose the right state to return to here */
@@ -607,7 +623,7 @@ int proxy_read_from_upstream( struct proxier* proxy, int state )
goto disconnect;
}
if ( ( proxy->req_hdr.type & REQUEST_MASK ) == REQUEST_READ ) {
if ( proxy->req_hdr.type == REQUEST_READ ) {
/* Get the read reply data too. */
proxy->rsp.size += proxy->req_hdr.len;
}
@@ -683,7 +699,7 @@ void proxy_session( struct proxier* proxy )
/* First action: Write hello to downstream */
nbd_hello_to_buf( (struct nbd_init_raw *) proxy->rsp.buf, proxy->upstream_size );
nbd_hello_to_buf( (struct nbd_init_raw *) proxy->rsp.buf, proxy->upstream_size, proxy->upstream_flags );
proxy->rsp.size = sizeof( struct nbd_init_raw );
proxy->rsp.needle = 0;
state = WRITE_TO_DOWNSTREAM;

5
src/proxy/proxy.h
View File

@@ -46,10 +46,13 @@ struct proxier {
/* This is the size we advertise to the downstream server */
uint64_t upstream_size;
/* These are the transmission flags sent as part of the handshake */
uint32_t upstream_flags;
/* We transform the raw request header into here */
struct nbd_request req_hdr;
/* We transform the raw reply header into here */
/* We transform the raw reply header into here */
struct nbd_reply rsp_hdr;
/* Used for our non-blocking negotiation with upstream. TODO: maybe use

37
src/server/client.c
View File

@@ -307,7 +307,11 @@ void client_write_init( struct client * client, uint64_t size )
memcpy( init.passwd, INIT_PASSWD, sizeof( init.passwd ) );
init.magic = INIT_MAGIC;
init.size = size;
memset( init.reserved, 0, 128 );
/* As more features are implemented, this is the place to advertise
* them.
*/
init.flags = FLAG_HAS_FLAGS | FLAG_SEND_FLUSH | FLAG_SEND_FUA;
memset( init.reserved, 0, 124 );
nbd_h2r_init( &init, &init_raw );
@@ -383,8 +387,8 @@ int client_request_needs_reply( struct client * client,
}
debug(
"request type=%"PRIu32", from=%"PRIu64", len=%"PRIu32", handle=0x%08X",
request.type, request.from, request.len, request.handle
"request type=%"PRIu16", flags=%"PRIu16", from=%"PRIu64", len=%"PRIu32", handle=0x%08X",
request.type, request.flags, request.from, request.len, request.handle
);
/* check it's not out of range */
@@ -411,7 +415,8 @@ int client_request_needs_reply( struct client * client,
debug("request disconnect");
client->disconnect = 1;
return 0;
case REQUEST_FLUSH:
break;
default:
fatal("Unknown request 0x%08X", request.type);
}
@@ -472,11 +477,13 @@ 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... */
// Only flush if FUA is set
if (request.flags & CMD_FLAG_FUA)
{
/* multiple of 4K page size */
uint64_t from_rounded = request.from & (!0xfff);
/* multiple of page size */
uint64_t from_rounded = request.from & (~(sysconf(_SC_PAGE_SIZE)-1));
uint64_t len_rounded = request.len + (request.from - from_rounded);
debug("Calling msync from=%"PRIu64", len=%"PRIu64"",from_rounded, len_rounded);
FATAL_IF_NEGATIVE(
msync( client->mapped + from_rounded,
@@ -488,6 +495,17 @@ void client_reply_to_write( struct client* client, struct nbd_request request )
client_write_reply( client, &request, 0);
}
void client_reply_to_flush( struct client* client, struct nbd_request request )
{
debug("request flush from=%"PRIu64", len=%"PRIu32", handle=0x%08X", request.from, request.len, request.handle);
ERROR_IF_NEGATIVE(
msync(client->mapped, client->mapped_size, MS_SYNC | MS_INVALIDATE),
"flush failed"
);
client_write_reply( client, &request, 0);
}
void client_reply( struct client* client, struct nbd_request request )
{
@@ -498,6 +516,9 @@ void client_reply( struct client* client, struct nbd_request request )
case REQUEST_WRITE:
client_reply_to_write( client, request );
break;
case REQUEST_FLUSH:
client_reply_to_flush( client, request );
break;
}
}
@@ -676,7 +697,7 @@ void* client_serve(void* client_uncast)
open_and_mmap(
client->serve->filename,
&client->fileno,
NULL,
&client->mapped_size,
(void**) &client->mapped
),
"Couldn't open/mmap file %s: %s", client->serve->filename, strerror( errno )

3
src/server/client.h
View File

@@ -3,6 +3,7 @@
#include <signal.h>
#include <time.h>
#include <inttypes.h>
/** CLIENT_HANDLER_TIMEOUT
* This is the length of time (in seconds) any request can be outstanding for.
@@ -31,6 +32,8 @@ struct client {
int fileno;
char* mapped;
uint64_t mapped_size;
struct self_pipe * stop_signal;
struct server* serve; /* FIXME: remove above duplication */

3
src/server/mirror.c
View File

@@ -293,7 +293,8 @@ int mirror_connect( struct mirror * mirror, uint64_t local_size )
if( FD_ISSET( mirror->client, &fds ) ){
uint64_t remote_size;
if ( socket_nbd_read_hello( mirror->client, &remote_size ) ) {
uint32_t remote_flags;
if ( socket_nbd_read_hello( mirror->client, &remote_size, &remote_flags ) ) {
if( remote_size == local_size ){
connected = 1;
mirror_set_state( mirror, MS_GO );

104
tests/acceptance/environment.rb
View File

@@ -1,39 +1,42 @@
# encoding: utf-8
require 'flexnbd'
require 'file_writer'
class Environment
attr_reader( :blocksize, :filename1, :filename2, :ip,
:port1, :port2, :nbd1, :nbd2, :file1, :file2 )
attr_reader(:blocksize, :filename1, :filename2, :ip,
:port1, :port2, :nbd1, :nbd2, :file1, :file2)
def initialize
@blocksize = 1024
@filename1 = "/tmp/.flexnbd.test.#{$$}.#{Time.now.to_i}.1"
@filename2 = "/tmp/.flexnbd.test.#{$$}.#{Time.now.to_i}.2"
@ip = "127.0.0.1"
@available_ports = [*40000..41000] - listening_ports
@filename1 = "/tmp/.flexnbd.test.#{$PROCESS_ID}.#{Time.now.to_i}.1"
@filename2 = "/tmp/.flexnbd.test.#{$PROCESS_ID}.#{Time.now.to_i}.2"
@ip = '127.0.0.1'
@available_ports = [*40_000..41_000] - listening_ports
@port1 = @available_ports.shift
@port2 = @available_ports.shift
@nbd1 = FlexNBD::FlexNBD.new("../../build/flexnbd", @ip, @port1)
@nbd2 = FlexNBD::FlexNBD.new("../../build/flexnbd", @ip, @port2)
@nbd1 = FlexNBD::FlexNBD.new('../../build/flexnbd', @ip, @port1)
@nbd2 = FlexNBD::FlexNBD.new('../../build/flexnbd', @ip, @port2)
@fake_pid = nil
end
def blocksize=(b)
raise RuntimeError, "Unable to change blocksize after files have been opened" if @file1 or @file2
@blocksize = b
end
def prefetch_proxy!
@nbd1.prefetch_proxy = true
@nbd2.prefetch_proxy = true
end
def proxy1(port=@port2)
def proxy1(port = @port2)
@nbd1.proxy(@ip, port)
end
def proxy2(port=@port1)
def proxy2(port = @port1)
@nbd2.proxy(@ip, port)
end
def serve1(*acl)
@nbd1.serve(@filename1, *acl)
end
@@ -42,29 +45,26 @@ class Environment
@nbd2.serve(@filename2, *acl)
end
def listen1( *acl )
@nbd1.listen( @filename1, *(acl.empty? ? @acl1: acl) )
def listen1(*acl)
@nbd1.listen(@filename1, *(acl.empty? ? @acl1 : acl))
end
def listen2( *acl )
@nbd2.listen( @filename2, *acl )
def listen2(*acl)
@nbd2.listen(@filename2, *acl)
end
def break1
@nbd1.break
end
def acl1( *acl )
@nbd1.acl( *acl )
def acl1(*acl)
@nbd1.acl(*acl)
end
def acl2( *acl )
@nbd2.acl( *acl )
def acl2(*acl)
@nbd2.acl(*acl)
end
def status1
@nbd1.status.first
end
@@ -73,23 +73,20 @@ class Environment
@nbd2.status.first
end
def mirror12
@nbd1.mirror( @nbd2.ip, @nbd2.port )
@nbd1.mirror(@nbd2.ip, @nbd2.port)
end
def mirror12_unchecked
@nbd1.mirror_unchecked( @nbd2.ip, @nbd2.port, nil, nil, 10 )
@nbd1.mirror_unchecked(@nbd2.ip, @nbd2.port, nil, nil, 10)
end
def mirror12_unlink
@nbd1.mirror_unlink( @nbd2.ip, @nbd2.port, 2 )
@nbd1.mirror_unlink(@nbd2.ip, @nbd2.port, 2)
end
def write1( data )
@nbd1.write( 0, data )
def write1(data)
@nbd1.write(0, data)
end
def writefile1(data)
@@ -100,63 +97,54 @@ class Environment
@file2 = FileWriter.new(@filename2, @blocksize).write(data)
end
def truncate1( size )
def truncate1(size)
system "truncate -s #{size} #{@filename1}"
end
def listening_ports
`netstat -ltn`.
split("\n").
map { |x| x.split(/\s+/) }[2..-1].
map { |l| l[3].split(":")[-1].to_i }
`netstat -ltn`
.split("\n")
.map { |x| x.split(/\s+/) }[2..-1]
.map { |l| l[3].split(':')[-1].to_i }
end
def cleanup
if @fake_pid
begin
Process.waitpid2( @fake_pid )
Process.waitpid2(@fake_pid)
rescue Errno::ESRCH
end
end
@nbd1.can_die(0)
@nbd1.kill
@nbd2.kill
[@filename1, @filename2].each do |f|
File.unlink(f) if File.exists?(f)
File.unlink(f) if File.exist?(f)
end
end
def run_fake( name, addr, port, sock=nil )
fakedir = File.join( File.dirname( __FILE__ ), "fakes" )
fakeglob = File.join( fakedir, name ) + "*"
fake = Dir[fakeglob].sort.find { |fn|
File.executable?( fn )
}
def run_fake(name, addr, port, sock = nil)
fakedir = File.join(File.dirname(__FILE__), 'fakes')
fakeglob = File.join(fakedir, name) + '*'
fake = Dir[fakeglob].sort.find do |fn|
File.executable?(fn)
end
raise "no fake executable at #{fakeglob}" unless fake
raise "no addr" unless addr
raise "no port" unless port
raise 'no addr' unless addr
raise 'no port' unless port
@fake_pid = fork do
exec [fake, addr, port, @nbd1.pid, sock].map{|x| x.to_s}.join(" ")
exec [fake, addr, port, @nbd1.pid, sock].map(&:to_s).join(' ')
end
sleep(0.5)
end
def fake_reports_success
_,status = Process.waitpid2( @fake_pid )
_, status = Process.waitpid2(@fake_pid)
@fake_pid = nil
status.success?
end
end # class Environment

19
tests/acceptance/fakes/dest/break_after_hello.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Open a server, accept a client, then cancel the migration by issuing
# a break command.
@@ -8,28 +6,27 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port, src_pid, sock = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
ctrl = UNIXSocket.open( sock )
ctrl = UNIXSocket.open(sock)
Process.kill("STOP", src_pid.to_i)
ctrl.write( "break\n" )
Process.kill('STOP', src_pid.to_i)
ctrl.write("break\n")
ctrl.close_write
client.write_hello
Process.kill("CONT", src_pid.to_i)
Process.kill('CONT', src_pid.to_i)
fail "Unexpected control response" unless
raise 'Unexpected control response' unless
ctrl.read =~ /0: mirror stopped/
client2 = nil
begin
client2 = server.accept( "Expected timeout" )
fail "Unexpected reconnection"
client2 = server.accept('Expected timeout')
raise 'Unexpected reconnection'
rescue Timeout::Error
# expected
end
client.close
exit(0)

19
tests/acceptance/fakes/dest/close_after_entrust_reply.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Receive a mirror, and disconnect after sending the entrust reply but
# before it can send the disconnect signal.
#
@@ -11,26 +9,25 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port, src_pid = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
client.write_hello
while (req = client.read_request; req[:type] == 1)
client.read_data( req[:len] )
client.write_reply( req[:handle] )
while req = client.read_request; req[:type] == 1
client.read_data(req[:len])
client.write_reply(req[:handle])
end
system "kill -STOP #{src_pid}"
client.write_reply( req[:handle] )
client.write_reply(req[:handle])
client.close
system "kill -CONT #{src_pid}"
sleep( 0.25 )
client2 = server.accept( "Timed out waiting for a reconnection" )
sleep(0.25)
client2 = server.accept('Timed out waiting for a reconnection')
client2.close
server.close
$stderr.puts "done"
warn 'done'
exit(0)

6
tests/acceptance/fakes/dest/close_after_hello.rb
View File

@@ -10,12 +10,12 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
client = server.accept( "Timed out waiting for a connection" )
server = FakeDest.new(addr, port)
client = server.accept('Timed out waiting for a connection')
client.write_hello
client.close
new_client = server.accept( "Timed out waiting for a reconnection" )
new_client = server.accept('Timed out waiting for a reconnection')
new_client.close
server.close

6
tests/acceptance/fakes/dest/close_after_write.rb
View File

@@ -11,13 +11,13 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
client = server.accept( "Timed out waiting for a connection" )
server = FakeDest.new(addr, port)
client = server.accept('Timed out waiting for a connection')
client.write_hello
client.read_request
client.close
new_client = server.accept( "Timed out waiting for a reconnection" )
new_client = server.accept('Timed out waiting for a reconnection')
new_client.close
server.close

12
tests/acceptance/fakes/dest/close_after_writes.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Open a server, accept a client, then we expect a single write
# followed by an entrust. However, we disconnect after the write so
# the entrust will fail. We don't expect a reconnection: the sender
@@ -10,16 +8,16 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port, src_pid = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
client.write_hello
req = client.read_request
data = client.read_data( req[:len] )
data = client.read_data(req[:len])
Process.kill("STOP", src_pid.to_i)
client.write_reply( req[:handle], 0 )
Process.kill('STOP', src_pid.to_i)
client.write_reply(req[:handle], 0)
client.close
Process.kill("CONT", src_pid.to_i)
Process.kill('CONT', src_pid.to_i)
exit(0)

9
tests/acceptance/fakes/dest/error_on_write.rb
View File

@@ -1,19 +1,16 @@
#!/usr/bin/env ruby
# encoding: utf-8
require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
client.write_hello
handle = client.read_request[:handle]
client.write_error( handle )
client.write_error(handle)
client2 = server.accept( "Timed out waiting for a reconnection" )
client2 = server.accept('Timed out waiting for a reconnection')
client.close
client2.close

10
tests/acceptance/fakes/dest/hang_after_connect.rb
View File

@@ -14,8 +14,8 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
client = server.accept( "Client didn't make a connection" )
server = FakeDest.new(addr, port)
client = server.accept("Client didn't make a connection")
# Sleep for one second past the timeout (a bit of slop in case ruby
# doesn't launch things quickly)
@@ -26,10 +26,10 @@ client.close
# Invert the sense of the timeout exception, since we *don't* want a
# connection attempt
begin
server.accept( "Expected timeout" )
fail "Unexpected reconnection"
server.accept('Expected timeout')
raise 'Unexpected reconnection'
rescue Timeout::Error
# expected
# expected
end
server.close

20
tests/acceptance/fakes/dest/hang_after_write.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Open a socket, say hello, receive a write, then sleep for >
# MS_REQUEST_LIMIT_SECS seconds. This should tell the source that the
# write has gone MIA, and we expect a reconnect.
@@ -9,24 +7,24 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
client1 = server.accept( server )
server = FakeDest.new(addr, port)
client1 = server.accept(server)
client1.write_hello
client1.read_request
t = Thread.start do
client2 = server.accept( "Timed out waiting for a reconnection",
FlexNBD::MS_REQUEST_LIMIT_SECS + 2 )
client2 = server.accept('Timed out waiting for a reconnection',
FlexNBD::MS_REQUEST_LIMIT_SECS + 2)
client2.close
end
sleep_time = if ENV.has_key?('FLEXNBD_MS_REQUEST_LIMIT_SECS')
ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS'].to_f
else
FlexNBD::MS_REQUEST_LIMIT_SECS
sleep_time = if ENV.key?('FLEXNBD_MS_REQUEST_LIMIT_SECS')
ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS'].to_f
else
FlexNBD::MS_REQUEST_LIMIT_SECS
end
sleep( sleep_time + 2.0 )
sleep(sleep_time + 2.0)
client1.close
t.join

12
tests/acceptance/fakes/dest/hello_wrong_magic.rb
View File

@@ -7,21 +7,21 @@ include FlexNBD
Thread.abort_on_exception
addr, port = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client1 = server.accept
# We don't expect a reconnection attempt.
t = Thread.new do
begin
client2 = server.accept( "Timed out waiting for a reconnection",
FlexNBD::MS_RETRY_DELAY_SECS + 1 )
fail "Unexpected reconnection"
client2 = server.accept('Timed out waiting for a reconnection',
FlexNBD::MS_RETRY_DELAY_SECS + 1)
raise 'Unexpected reconnection'
rescue Timeout::Error
#expected
# expected
end
end
client1.write_hello( :magic => :wrong )
client1.write_hello(magic: :wrong)
t.join

12
tests/acceptance/fakes/dest/hello_wrong_size.rb
View File

@@ -9,7 +9,7 @@ include FlexNBD
Thread.abort_on_exception = true
addr, port = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
t = Thread.new do
@@ -18,21 +18,21 @@ t = Thread.new do
# so it makes no sense to continue. This means we have to invert the
# sense of the exception.
begin
client2 = server.accept( "Timed out waiting for a reconnection",
FlexNBD::MS_RETRY_DELAY_SECS + 1 )
client2 = server.accept('Timed out waiting for a reconnection',
FlexNBD::MS_RETRY_DELAY_SECS + 1)
client2.close
fail "Unexpected reconnection."
raise 'Unexpected reconnection.'
rescue Timeout::Error
end
end
client.write_hello( :size => :wrong )
client.write_hello(size: :wrong)
t.join
# Now check that the source closed the first socket (yes, this was an
# actual bug)
fail "Didn't close socket" unless client.disconnected?
raise "Didn't close socket" unless client.disconnected?
exit 0

6
tests/acceptance/fakes/dest/reject_acl.rb
View File

@@ -7,18 +7,16 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
server.accept.close
begin
server.accept
fail "Unexpected reconnection"
raise 'Unexpected reconnection'
rescue Timeout::Error
# expected
end
server.close
exit(0)

4
tests/acceptance/fakes/dest/sigterm_after_hello.rb
View File

@@ -8,8 +8,8 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port, pid = *ARGV
server = FakeDest.new( addr, port )
client = server.accept( "Timed out waiting for a connection" )
server = FakeDest.new(addr, port)
client = server.accept('Timed out waiting for a connection')
client.write_hello
Process.kill(15, pid.to_i)

8
tests/acceptance/fakes/dest/write_wrong_magic.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Accept a connection, write hello, wait for a write request, read the
# data, then write back a reply with a bad magic field. We then
# expect a reconnect.
@@ -9,13 +7,13 @@ require 'flexnbd/fake_dest'
include FlexNBD
addr, port = *ARGV
server = FakeDest.new( addr, port )
server = FakeDest.new(addr, port)
client = server.accept
client.write_hello
req = client.read_request
client.read_data( req[:len] )
client.write_reply( req[:handle], 0, :magic => :wrong )
client.read_data(req[:len])
client.write_reply(req[:handle], 0, magic: :wrong)
client2 = server.accept
client.close

14
tests/acceptance/fakes/source/close_after_connect.rb
View File

@@ -11,13 +11,13 @@ include FlexNBD
addr, port = *ARGV
FakeSource.new( addr, port, "Failed to connect" ).close
# Sleep to be sure we don't try to connect too soon. That wouldn't
# be a problem for the destination, but it would prevent us from
# determining success or failure here in the case where we try to
# reconnect before the destination has tidied up after the first
# thread went away.
FakeSource.new(addr, port, 'Failed to connect').close
# Sleep to be sure we don't try to connect too soon. That wouldn't
# be a problem for the destination, but it would prevent us from
# determining success or failure here in the case where we try to
# reconnect before the destination has tidied up after the first
# thread went away.
sleep(0.5)
FakeSource.new( addr, port, "Failed to reconnect" ).close
FakeSource.new(addr, port, 'Failed to reconnect').close
exit 0

11
tests/acceptance/fakes/source/close_after_entrust.rb
View File

@@ -11,10 +11,10 @@ include FlexNBD
addr, port, srv_pid = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.read_hello
client.write_write_request( 0, 8 )
client.write_data( "12345678" )
client.write_write_request(0, 8)
client.write_data('12345678')
# Use system "kill" rather than Process.kill because Process.kill
# doesn't seem to work
@@ -25,12 +25,11 @@ client.close
system "kill -CONT #{srv_pid}"
sleep(0.25)
begin
client2 = FakeSource.new( addr, port, "Expected timeout" )
fail "Unexpected reconnection"
client2 = FakeSource.new(addr, port, 'Expected timeout')
raise 'Unexpected reconnection'
rescue Timeout::Error
# expected
end

12
tests/acceptance/fakes/source/close_after_entrust_reply.rb
View File

@@ -10,10 +10,10 @@ include FlexNBD
addr, port, srv_pid = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.read_hello
client.write_write_request( 0, 8 )
client.write_data( "12345678" )
client.write_write_request(0, 8)
client.write_data('12345678')
client.write_entrust_request
client.read_response
@@ -21,13 +21,11 @@ client.close
sleep(0.25)
begin
client2 = FakeSource.new( addr, port, "Expected timeout" )
fail "Unexpected reconnection"
client2 = FakeSource.new(addr, port, 'Expected timeout')
raise 'Unexpected reconnection'
rescue Timeout::Error
# expected
end
exit(0)

5
tests/acceptance/fakes/source/close_after_hello.rb
View File

@@ -12,13 +12,12 @@ include FlexNBD
addr, port = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting." )
client = FakeSource.new(addr, port, 'Timed out connecting.')
client.read_hello
client.close
sleep(0.2)
FakeSource.new( addr, port, "Timed out reconnecting." )
FakeSource.new(addr, port, 'Timed out reconnecting.')
exit(0)

8
tests/acceptance/fakes/source/close_after_write.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# We connect, pause the server, issue a write request, disconnect,
# then cont the server. This ensures that our disconnect happens
# while the server is trying to read the write data.
@@ -10,11 +8,11 @@ include FlexNBD
addr, port, srv_pid = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.read_hello
system "kill -STOP #{srv_pid}"
client.write_write_request( 0, 8 )
client.write_write_request(0, 8)
client.close
system "kill -CONT #{srv_pid}"
@@ -24,7 +22,7 @@ system "kill -CONT #{srv_pid}"
sleep(0.25)
# ...and can we reconnect?
client2 = FakeSource.new( addr, port, "Timed out connecting" )
client2 = FakeSource.new(addr, port, 'Timed out connecting')
client2.close
exit(0)

10
tests/acceptance/fakes/source/close_after_write_data.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# We connect, pause the server, issue a write request, send data,
# disconnect, then cont the server. This ensures that our disconnect
# happens before the server can try to write the reply.
@@ -10,13 +8,13 @@ include FlexNBD
addr, port, srv_pid = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.read_hello
system "kill -STOP #{srv_pid}"
client.write_write_request( 0, 8 )
client.write_data( "12345678" )
client.write_write_request(0, 8)
client.write_data('12345678')
client.close
system "kill -CONT #{srv_pid}"
@@ -27,7 +25,7 @@ system "kill -CONT #{srv_pid}"
sleep(0.25)
# ...and can we reconnect?
client2 = FakeSource.new( addr, port, "Timed out reconnecting" )
client2 = FakeSource.new(addr, port, 'Timed out reconnecting')
client2.close
exit(0)

7
tests/acceptance/fakes/source/close_mid_read.rb
View File

@@ -8,10 +8,9 @@ include FlexNBD
addr, port, srv_pid, newaddr, newport = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client.write_read_request( 0, 8 )
client.read_raw( 4 )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.write_read_request(0, 8)
client.read_raw(4)
client.close
exit(0)

4
tests/acceptance/fakes/source/connect_during_hello.rb
View File

@@ -10,9 +10,9 @@ include FlexNBD
addr, port = *ARGV
client1 = FakeSource.new( addr, port, "Timed out connecting" )
client1 = FakeSource.new(addr, port, 'Timed out connecting')
sleep(0.25)
client2 = FakeSource.new( addr, port, "Timed out connecting a second time" )
client2 = FakeSource.new(addr, port, 'Timed out connecting a second time')
# This is the expected source crashing after connect
client1.close

7
tests/acceptance/fakes/source/connect_from_banned_ip.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# We connect from a local address which should be blocked, sleep for a
# bit, then try to read from the socket. We should get an instant EOF
# as we've been cut off by the destination.
@@ -11,10 +9,9 @@ include FlexNBD
addr, port = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting", "127.0.0.6" )
sleep( 0.25 )
client = FakeSource.new(addr, port, 'Timed out connecting', '127.0.0.6')
sleep(0.25)
rsp = client.disconnected? ? 0 : 1
client.close
exit(rsp)

12
tests/acceptance/fakes/source/hang_after_hello.rb
View File

@@ -7,10 +7,10 @@
# listening for an incoming migration.
addr, port = *ARGV
require "flexnbd/fake_source"
require 'flexnbd/fake_source'
include FlexNBD
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
client.read_hello
# Now we do two things:
@@ -24,16 +24,16 @@ client.read_hello
kidpid = fork do
client.close
new_client = nil
sleep( FlexNBD::CLIENT_MAX_WAIT_SECS + 1 )
new_client = FakeSource.new( addr, port, "Timed out reconnecting." )
sleep(FlexNBD::CLIENT_MAX_WAIT_SECS + 1)
new_client = FakeSource.new(addr, port, 'Timed out reconnecting.')
new_client.read_hello
exit 0
end
# Sleep for longer than the child, to give the flexnbd process a bit
# of slop
sleep( FlexNBD::CLIENT_MAX_WAIT_SECS + 3 )
sleep(FlexNBD::CLIENT_MAX_WAIT_SECS + 3)
client.close
_,status = Process.waitpid2( kidpid )
_, status = Process.waitpid2(kidpid)
exit status.exitstatus

4
tests/acceptance/fakes/source/sigterm_after_hello.rb
View File

@@ -9,10 +9,10 @@ include FlexNBD
addr, port, pid = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting." )
client = FakeSource.new(addr, port, 'Timed out connecting.')
client.read_hello
Process.kill( "TERM", pid.to_i )
Process.kill('TERM', pid.to_i)
sleep(0.2)
client.close

7
tests/acceptance/fakes/source/successful_transfer.rb
View File

@@ -9,10 +9,9 @@ include FlexNBD
addr, port, srv_pid, newaddr, newport = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client.send_mirror()
client = FakeSource.new(addr, port, 'Timed out connecting')
client.send_mirror
sleep(1)
exit( 0 )
exit(0)

18
tests/acceptance/fakes/source/write_out_of_range.rb
View File

@@ -1,6 +1,4 @@
#!/usr/bin/env ruby
# encoding: utf-8
# Connect, read the hello then make a write request with an impossible
# (from,len) pair. We expect an error response, and not to be
# disconnected.
@@ -13,20 +11,20 @@ include FlexNBD
addr, port = *ARGV
client = FakeSource.new( addr, port, "Timed out connecting" )
client = FakeSource.new(addr, port, 'Timed out connecting')
hello = client.read_hello
client.write_write_request( hello[:size]+1, 32, "myhandle" )
client.write_data("1"*32)
client.write_write_request(hello[:size] + 1, 32, 'myhandle')
client.write_data('1' * 32)
response = client.read_response
fail "Not an error" if response[:error] == 0
fail "Wrong handle" unless "myhandle" == response[:handle]
raise 'Not an error' if response[:error] == 0
raise 'Wrong handle' unless response[:handle] == 'myhandle'
client.write_write_request( 0, 32 )
client.write_data( "2"*32 )
client.write_write_request(0, 32)
client.write_data('2' * 32)
success_response = client.read_response
fail "Second write failed" unless success_response[:error] == 0
raise 'Second write failed' unless success_response[:error] == 0
client.close
exit(0)

87
tests/acceptance/file_writer.rb
View File

@@ -3,13 +3,13 @@
#
class FileWriter
def initialize(filename, blocksize)
@fh = File.open(filename, "w+")
@fh = File.open(filename, 'w+')
@blocksize = blocksize
@pattern = ""
@pattern = ''
end
def size
@blocksize * @pattern.split("").size
@blocksize * @pattern.split('').size
end
# We write in fixed block sizes, given by "blocksize"
@@ -20,8 +20,8 @@ class FileWriter
def write(data)
@pattern += data
data.split("").each do |code|
if code == "_"
data.split('').each do |code|
if code == '_'
@fh.seek(@blocksize, IO::SEEK_CUR)
else
@fh.write(data(code))
@@ -31,15 +31,14 @@ class FileWriter
self
end
# Returns what the data ought to be at the given offset and length
#
def read_original( off, len )
patterns = @pattern.split( "" )
patterns.zip( (0...patterns.length).to_a ).
map { |blk, blk_off|
def read_original(off, len)
patterns = @pattern.split('')
patterns.zip((0...patterns.length).to_a)
.map do |blk, blk_off|
data(blk, blk_off)
}.join[off...(off+len)]
end.join[off...(off + len)]
end
# Read what's actually in the file
@@ -60,68 +59,66 @@ class FileWriter
protected
def data(code, at=@fh.tell)
def data(code, at = @fh.tell)
case code
when "0", "_"
"\0" * @blocksize
when "X"
"X" * @blocksize
when "f"
r = ""
(@blocksize/4).times do
r += [at].pack("I")
at += 4
end
r
else
raise "Unknown character '#{block}'"
when '0', '_'
"\0" * @blocksize
when 'X'
'X' * @blocksize
when 'f'
r = ''
(@blocksize / 4).times do
r += [at].pack('I')
at += 4
end
r
else
raise "Unknown character '#{block}'"
end
end
end
if __FILE__==$0
if $PROGRAM_NAME == __FILE__
require 'tempfile'
require 'test/unit'
class FileWriterTest < Test::Unit::TestCase
def test_read_original_zeros
Tempfile.open("test_read_original_zeros") do |tempfile|
Tempfile.open('test_read_original_zeros') do |tempfile|
tempfile.close
file = FileWriter.new( tempfile.path, 4096 )
file.write( "0" )
assert_equal file.read( 0, 4096 ), file.read_original( 0, 4096 )
assert( file.untouched?(0,4096) , "Untouched file was touched." )
file = FileWriter.new(tempfile.path, 4096)
file.write('0')
assert_equal file.read(0, 4096), file.read_original(0, 4096)
assert(file.untouched?(0, 4096), 'Untouched file was touched.')
end
end
def test_read_original_offsets
Tempfile.open("test_read_original_offsets") do |tempfile|
Tempfile.open('test_read_original_offsets') do |tempfile|
tempfile.close
file = FileWriter.new( tempfile.path, 4096 )
file.write( "f" )
assert_equal file.read( 0, 4096 ), file.read_original( 0, 4096 )
assert( file.untouched?(0,4096) , "Untouched file was touched." )
file = FileWriter.new(tempfile.path, 4096)
file.write('f')
assert_equal file.read(0, 4096), file.read_original(0, 4096)
assert(file.untouched?(0, 4096), 'Untouched file was touched.')
end
end
def test_file_size
Tempfile.open("test_file_size") do |tempfile|
Tempfile.open('test_file_size') do |tempfile|
tempfile.close
file = FileWriter.new( tempfile.path, 4096 )
file.write( "f" )
assert_equal 4096, File.stat( tempfile.path ).size
file = FileWriter.new(tempfile.path, 4096)
file.write('f')
assert_equal 4096, File.stat(tempfile.path).size
end
end
def test_read_original_size
Tempfile.open("test_read_original_offsets") do |tempfile|
Tempfile.open('test_read_original_offsets') do |tempfile|
tempfile.close
file = FileWriter.new( tempfile.path, 4)
file.write( "f"*4 )
file = FileWriter.new(tempfile.path, 4)
file.write('f' * 4)
assert_equal 4, file.read_original(0, 4).length
end
end
end
end

328
tests/acceptance/flexnbd.rb
View File

@@ -4,28 +4,26 @@ require 'open3'
require 'timeout'
require 'rexml/document'
require 'rexml/streamlistener'
require 'English'
Thread.abort_on_exception = true
class Executor
attr_reader :pid
def run( cmd )
@pid = fork do exec cmd end
def run(cmd)
@pid = fork { exec cmd }
end
end # class Executor
class ValgrindExecutor
attr_reader :pid
def run( cmd )
@pid = fork do exec "valgrind --track-origins=yes --suppressions=custom.supp #{cmd}" end
def run(cmd)
@pid = fork { exec "valgrind --track-origins=yes --suppressions=custom.supp #{cmd}" }
end
end # class ValgrindExecutor
class ValgrindKillingExecutor
attr_reader :pid
@@ -33,10 +31,10 @@ class ValgrindKillingExecutor
attr_accessor :what, :kind, :pid
attr_reader :backtrace
def initialize
@backtrace=[]
@what = ""
@kind = ""
@pid = ""
@backtrace = []
@what = ''
@kind = ''
@pid = ''
end
def add_frame
@@ -56,115 +54,104 @@ class ValgrindKillingExecutor
end
def to_s
([@what + " (#{@kind}) in #{@pid}"] + @backtrace.map{|h| "#{h[:file]}:#{h[:line]} #{h[:fn]}" }).join("\n")
([@what + " (#{@kind}) in #{@pid}"] + @backtrace.map { |h| "#{h[:file]}:#{h[:line]} #{h[:fn]}" }).join("\n")
end
end # class Error
class ErrorListener
include REXML::StreamListener
def initialize( killer )
def initialize(killer)
@killer = killer
@error = Error.new
@found = false
end
def text( text )
def text(text)
@text = text
end
def tag_start(tag, attrs)
def tag_start(tag, _attrs)
case tag.to_s
when "error"
when 'error'
@found = true
when "frame"
when 'frame'
@error.add_frame
end
end
def tag_end(tag)
case tag.to_s
when "what"
when 'what'
@error.what = @text if @found
@text = ""
when "kind"
@text = ''
when 'kind'
@error.kind = @text if @found
when "file"
@error.add_file( @text ) if @found
when "fn"
@error.add_fn( @text ) if @found
when "line"
@error.add_line( @text ) if @found
when "error", "stack"
if @found
@killer.call( @error )
end
when "pid"
@error.pid=@text
when 'file'
@error.add_file(@text) if @found
when 'fn'
@error.add_fn(@text) if @found
when 'line'
@error.add_line(@text) if @found
when 'error', 'stack'
@killer.call(@error) if @found
when 'pid'
@error.pid = @text
end
end
end # class ErrorListener
class DebugErrorListener < ErrorListener
def text( txt )
def text(txt)
print txt
super( txt )
super(txt)
end
def tag_start( tag, attrs )
def tag_start(tag, attrs)
print "<#{tag}>"
super( tag, attrs )
super(tag, attrs)
end
def tag_end( tag )
def tag_end(tag)
print "</#{tag}>"
super( tag )
super(tag)
end
end
def initialize
@pid = nil
end
def run( cmd )
def run(cmd)
@io_r, io_w = IO.pipe
@pid = fork do exec( "valgrind --suppressions=custom.supp --xml=yes --xml-fd=#{io_w.fileno} " + cmd ) end
launch_watch_thread( @pid, @io_r )
@pid = fork { exec("valgrind --suppressions=custom.supp --xml=yes --xml-fd=#{io_w.fileno} " + cmd) }
launch_watch_thread(@pid, @io_r)
@pid
end
def call( err )
$stderr.puts "*"*72
$stderr.puts "* Valgrind error spotted:"
$stderr.puts err.to_s.split("\n").map{|s| " #{s}"}
$stderr.puts "*"*72
Process.kill( "KILL", @pid )
def call(err)
warn '*' * 72
warn '* Valgrind error spotted:'
warn err.to_s.split("\n").map { |s| " #{s}" }
warn '*' * 72
Process.kill('KILL', @pid)
exit(1)
end
private
def pick_listener
ENV['DEBUG'] ? DebugErrorListener : ErrorListener
end
def launch_watch_thread(pid, io_r)
def launch_watch_thread(_pid, io_r)
Thread.start do
io_source = REXML::IOSource.new( io_r )
listener = pick_listener.new( self )
REXML::Document.parse_stream( io_source, listener )
io_source = REXML::IOSource.new(io_r)
listener = pick_listener.new(self)
REXML::Document.parse_stream(io_source, listener)
end
end
end # class ValgrindExecutor
module FlexNBD
# Noddy test class to exercise FlexNBD from the outside for testing.
#
@@ -173,7 +160,7 @@ module FlexNBD
class << self
def counter
Dir['tmp/*'].select{|f| File.file?(f)}.length + 1
Dir['tmp/*'].select { |f| File.file?(f) }.length + 1
end
end
@@ -189,19 +176,18 @@ module FlexNBD
end
end
def build_debug_opt
if @do_debug
"--verbose"
'--verbose'
else
"--quiet"
'--quiet'
end
end
attr_accessor :prefetch_proxy
def initialize( bin, ip, port )
@bin = bin
def initialize(bin, ip, port)
@bin = bin
@do_debug = ENV['DEBUG']
@debug = build_debug_opt
raise "#{bin} not executable" unless File.executable?(bin)
@@ -209,21 +195,20 @@ module FlexNBD
@ctrl = "/tmp/.flexnbd.ctrl.#{Time.now.to_i}.#{rand}"
@ip = ip
@port = port
@pid = @wait_thread = nil
@kill = []
@prefetch_proxy = false
end
def debug?
!!@do_debug
end
def debug( msg )
$stderr.puts msg if debug?
def debug(msg)
warn msg if debug?
end
def serve_cmd( file, acl )
def serve_cmd(file, acl)
"#{bin} serve "\
"--addr #{ip} "\
"--port #{port} "\
@@ -233,8 +218,7 @@ module FlexNBD
"#{acl.join(' ')}"
end
def listen_cmd( file, acl )
def listen_cmd(file, acl)
"#{bin} listen "\
"--addr #{ip} "\
"--port #{port} "\
@@ -244,18 +228,17 @@ module FlexNBD
"#{acl.join(' ')}"
end
def proxy_cmd( connect_ip, connect_port )
def proxy_cmd(connect_ip, connect_port)
"#{bin}-proxy "\
"--addr #{ip} "\
"--port #{port} "\
"--conn-addr #{connect_ip} "\
"--conn-port #{connect_port} "\
"#{prefetch_proxy ? "--cache " : ""}"\
"#{prefetch_proxy ? '--cache ' : ''}"\
"#{@debug}"
end
def read_cmd( offset, length )
def read_cmd(offset, length)
"#{bin} read "\
"--addr #{ip} "\
"--port #{port} "\
@@ -264,8 +247,7 @@ module FlexNBD
"--size #{length}"
end
def write_cmd( offset, data )
def write_cmd(offset, data)
"#{bin} write "\
"--addr #{ip} "\
"--port #{port} "\
@@ -274,30 +256,29 @@ module FlexNBD
"--size #{data.length}"
end
def base_mirror_opts( dest_ip, dest_port )
def base_mirror_opts(dest_ip, dest_port)
"--addr #{dest_ip} "\
"--port #{dest_port} "\
"--sock #{ctrl} "\
end
def unlink_mirror_opts( dest_ip, dest_port )
"#{base_mirror_opts( dest_ip, dest_port )} "\
"--unlink "
def unlink_mirror_opts(dest_ip, dest_port)
"#{base_mirror_opts(dest_ip, dest_port)} "\
'--unlink '
end
def base_mirror_cmd( opts )
def base_mirror_cmd(opts)
"#{@bin} mirror "\
"#{opts} "\
"#{@debug}"
end
def mirror_cmd(dest_ip, dest_port)
base_mirror_cmd( base_mirror_opts( dest_ip, dest_port ) )
base_mirror_cmd(base_mirror_opts(dest_ip, dest_port))
end
def mirror_unlink_cmd( dest_ip, dest_port )
base_mirror_cmd( unlink_mirror_opts( dest_ip, dest_port ) )
def mirror_unlink_cmd(dest_ip, dest_port)
base_mirror_cmd(unlink_mirror_opts(dest_ip, dest_port))
end
def break_cmd
@@ -312,58 +293,64 @@ module FlexNBD
"#{@debug}"
end
def acl_cmd( *acl )
def acl_cmd(*acl)
"#{@bin} acl " \
"--sock #{ctrl} "\
"#{@debug} "\
"#{acl.join " "}"
"#{acl.join ' '}"
end
def run_serve_cmd(cmd)
File.unlink(ctrl) if File.exists?(ctrl)
debug( cmd )
File.unlink(ctrl) if File.exist?(ctrl)
debug(cmd)
@pid = @executor.run( cmd )
@pid = @executor.run(cmd)
while !File.socket?(ctrl)
until File.socket?(ctrl)
pid, status = Process.wait2(@pid, Process::WNOHANG)
raise "server did not start (#{cmd})" if pid
sleep 0.1
end
start_wait_thread( @pid )
start_wait_thread(@pid)
at_exit { kill }
end
private :run_serve_cmd
def serve( file, *acl)
cmd = serve_cmd( file, acl )
run_serve_cmd( cmd )
sleep( 0.2 ) until File.exists?( ctrl )
def serve(file, *acl)
cmd = serve_cmd(file, acl)
run_serve_cmd(cmd)
sleep(0.2) until File.exist?(ctrl)
end
def listen(file, *acl)
run_serve_cmd( listen_cmd( file, acl ) )
run_serve_cmd(listen_cmd(file, acl))
end
def tcp_server_open?
# raises if the other side doesn't accept()
sock = TCPSocket.new(ip, port) rescue nil
sock = begin
TCPSocket.new(ip, port)
rescue StandardError
nil
end
success = !!sock
( sock.close rescue nil) if sock
if sock
(begin
sock.close
rescue StandardError
nil
end)
end
success
end
def proxy( connect_ip, connect_port )
cmd = proxy_cmd( connect_ip, connect_port )
debug( cmd )
def proxy(connect_ip, connect_port)
cmd = proxy_cmd(connect_ip, connect_port)
debug(cmd)
@pid = @executor.run( cmd )
@pid = @executor.run(cmd)
until tcp_server_open?
pid, status = Process.wait2(@pid, Process::WNOHANG)
@@ -371,31 +358,29 @@ module FlexNBD
sleep 0.1
end
start_wait_thread( @pid )
start_wait_thread(@pid)
at_exit { kill }
end
def start_wait_thread( pid )
def start_wait_thread(pid)
@wait_thread = Thread.start do
_, status = Process.waitpid2( pid )
_, status = Process.waitpid2(pid)
if @kill
if status.signaled?
fail "flexnbd quit with a bad signal: #{status.inspect}" unless
raise "flexnbd quit with a bad signal: #{status.inspect}" unless
@kill.include? status.termsig
else
fail "flexnbd quit with a bad status: #{status.inspect}" unless
raise "flexnbd quit with a bad status: #{status.inspect}" unless
@kill.include? status.exitstatus
end
else
$stderr.puts "flexnbd #{self.pid} quit"
fail "flexnbd #{self.pid} quit early with status #{status.to_i}"
warn "flexnbd #{self.pid} quit"
raise "flexnbd #{self.pid} quit early with status #{status.to_i}"
end
end
end
def can_die(*status)
status = [0] if status.empty?
@kill += status
@@ -407,7 +392,7 @@ module FlexNBD
can_die(1)
if @pid
begin
Process.kill("INT", @pid)
Process.kill('INT', @pid)
rescue Errno::ESRCH => e
# already dead. Presumably this means it went away after a
# can_die() call.
@@ -417,63 +402,60 @@ module FlexNBD
end
def read(offset, length)
cmd = read_cmd( offset, length )
debug( cmd )
cmd = read_cmd(offset, length)
debug(cmd)
IO.popen(cmd) do |fh|
return fh.read
end
raise IOError.new "NBD read failed" unless $?.success?
raise IOError, 'NBD read failed' unless $CHILD_STATUS.success?
out
end
def write(offset, data)
cmd = write_cmd( offset, data )
debug( cmd )
cmd = write_cmd(offset, data)
debug(cmd)
IO.popen(cmd, "w") do |fh|
IO.popen(cmd, 'w') do |fh|
fh.write(data)
end
raise IOError.new "NBD write failed" unless $?.success?
raise IOError, 'NBD write failed' unless $CHILD_STATUS.success?
nil
end
def join
@wait_thread.join
end
def mirror_unchecked(dest_ip, dest_port, _bandwidth = nil, _action = nil, timeout = nil)
cmd = mirror_cmd(dest_ip, dest_port)
debug(cmd)
def mirror_unchecked( dest_ip, dest_port, bandwidth=nil, action=nil, timeout=nil )
cmd = mirror_cmd( dest_ip, dest_port)
debug( cmd )
maybe_timeout( cmd, timeout )
maybe_timeout(cmd, timeout)
end
def mirror_unlink(dest_ip, dest_port, timeout = nil)
cmd = mirror_unlink_cmd(dest_ip, dest_port)
debug(cmd)
def mirror_unlink( dest_ip, dest_port, timeout=nil )
cmd = mirror_unlink_cmd( dest_ip, dest_port )
debug( cmd )
maybe_timeout( cmd, timeout )
maybe_timeout(cmd, timeout)
end
def maybe_timeout(cmd, timeout=nil )
stdout, stderr = "",""
def maybe_timeout(cmd, timeout = nil)
stdout = ''
stderr = ''
stat = nil
run = Proc.new do
run = proc do
# Ruby 1.9 changed the popen3 api. instead of 3 args, the block
# gets 4. Not only that, but it no longer sets $?, so we have to
# go elsewhere for the process' exit status.
Open3.popen3( cmd ) do |io_in, io_out, io_err, maybe_thr|
Open3.popen3(cmd) do |io_in, io_out, io_err, maybe_thr|
io_in.close
stdout.replace io_out.read
stderr.replace io_err.read
stat = maybe_thr.value if maybe_thr
end
stat ||= $?
stat ||= $CHILD_STATUS
end
if timeout
@@ -485,85 +467,73 @@ module FlexNBD
[stdout, stderr, stat]
end
def mirror(dest_ip, dest_port, bandwidth=nil, action=nil)
stdout, stderr, status = mirror_unchecked( dest_ip, dest_port, bandwidth, action )
raise IOError.new( "Migrate command failed\n" + stderr) unless status.success?
def mirror(dest_ip, dest_port, bandwidth = nil, action = nil)
stdout, stderr, status = mirror_unchecked(dest_ip, dest_port, bandwidth, action)
raise IOError, "Migrate command failed\n" + stderr unless status.success?
stdout
end
def break(timeout=nil)
def break(timeout = nil)
cmd = break_cmd
debug( cmd )
debug(cmd)
maybe_timeout( cmd, timeout )
maybe_timeout(cmd, timeout)
end
def acl(*acl)
cmd = acl_cmd( *acl )
debug( cmd )
cmd = acl_cmd(*acl)
debug(cmd)
maybe_timeout( cmd, 2 )
maybe_timeout(cmd, 2)
end
def status(timeout = nil)
cmd = status_cmd
debug(cmd)
def status( timeout = nil )
cmd = status_cmd()
debug( cmd )
o,e = maybe_timeout( cmd, timeout )
o, e = maybe_timeout(cmd, timeout)
[parse_status(o), e]
end
def launched?
!!@pid
end
def paused
Process.kill( "STOP", @pid )
Process.kill('STOP', @pid)
yield
ensure
Process.kill( "CONT", @pid )
Process.kill('CONT', @pid)
end
protected
def control_command(*args)
raise "Server not running" unless @pid
raise 'Server not running' unless @pid
args = args.compact
UNIXSocket.open(@ctrl) do |u|
u.write(args.join("\n") + "\n")
code, message = u.readline.split(": ", 2)
code, message = u.readline.split(': ', 2)
return [code, message]
end
end
def parse_status( status )
def parse_status(status)
hsh = {}
status.split(" ").each do |part|
status.split(' ').each do |part|
next if part.strip.empty?
a,b = part.split("=")
a, b = part.split('=')
b.strip!
b = true if b == "true"
b = false if b == "false"
b = true if b == 'true'
b = false if b == 'false'
hsh[a.strip] = b
end
hsh
end
end
end

28
tests/acceptance/flexnbd/constants.rb
View File

@@ -1,10 +1,7 @@
# encoding: utf-8
module FlexNBD
def self.binary( str )
def self.binary(str)
if str.respond_to? :force_encoding
str.force_encoding "ASCII-8BIT"
str.force_encoding 'ASCII-8BIT'
else
str
end
@@ -13,36 +10,33 @@ module FlexNBD
# eeevil is his one and only name...
def self.read_constants
parents = []
current = File.expand_path(".")
while current != "/"
current = File.expand_path('.')
while current != '/'
parents << current
current = File.expand_path( File.join( current, ".." ) )
current = File.expand_path(File.join(current, '..'))
end
source_root = parents.find do |dirname|
File.directory?( File.join( dirname, "src" ) )
File.directory?(File.join(dirname, 'src'))
end
fail "No source root!" unless source_root
raise 'No source root!' unless source_root
headers = Dir[File.join( source_root, "src", "{common,proxy,server}","*.h" ) ]
headers = Dir[File.join(source_root, 'src', '{common,proxy,server}', '*.h')]
headers.each do |header_filename|
txt_lines = File.readlines( header_filename )
txt_lines = File.readlines(header_filename)
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 const_defined?( $1 )
const_set(Regexp.last_match(1), Regexp.last_match(2).to_i) unless const_defined?(Regexp.last_match(1))
end
end
end
end
read_constants()
read_constants
REQUEST_MAGIC = binary("\x25\x60\x95\x13") unless defined?(REQUEST_MAGIC)
REPLY_MAGIC = binary("\x67\x44\x66\x98") unless defined?(REPLY_MAGIC)
end # module FlexNBD

116
tests/acceptance/flexnbd/fake_dest.rb
View File

@@ -1,5 +1,3 @@
# encoding: utf-8
require 'socket'
require 'timeout'
@@ -7,114 +5,104 @@ require 'flexnbd/constants'
module FlexNBD
class FakeDest
class Client
def initialize( sock )
def initialize(sock)
@sock = sock
end
def write_hello( opts = {} )
@sock.write( "NBDMAGIC" )
def write_hello(opts = {})
@sock.write('NBDMAGIC')
if opts[:magic] == :wrong
write_rand( @sock, 8 )
write_rand(@sock, 8)
else
@sock.write( "\x00\x00\x42\x02\x81\x86\x12\x53" )
@sock.write("\x00\x00\x42\x02\x81\x86\x12\x53")
end
if opts[:size] == :wrong
write_rand( @sock, 8 )
write_rand(@sock, 8)
else
@sock.write( "\x00\x00\x00\x00\x00\x00\x10\x00" )
@sock.write("\x00\x00\x00\x00\x00\x00\x10\x00")
end
@sock.write( "\x00" * 128 )
@sock.write("\x00" * 128)
end
def write_rand( sock, len )
len.times do sock.write( rand(256).chr ) end
def write_rand(sock, len)
len.times { sock.write(rand(256).chr) }
end
def read_request()
def read_request
req = @sock.read(28)
magic_s = req[0 ... 4 ]
type_s = req[4 ... 8 ]
handle_s = req[8 ... 16]
from_s = req[16 ... 24]
len_s = req[24 ... 28]
magic_s = req[0...4]
type_s = req[4...8]
handle_s = req[8...16]
from_s = req[16...24]
len_s = req[24...28]
{
:magic => magic_s,
:type => type_s.unpack("N").first,
:handle => handle_s,
:from => self.class.parse_be64( from_s ),
:len => len_s.unpack( "N").first
magic: magic_s,
type: type_s.unpack('N').first,
handle: handle_s,
from: self.class.parse_be64(from_s),
len: len_s.unpack('N').first
}
end
def write_error( handle )
write_reply( handle, 1 )
def write_error(handle)
write_reply(handle, 1)
end
def disconnected?
begin
Timeout.timeout(2) do
@sock.read(1) == nil
end
rescue Timeout::Error
return false
Timeout.timeout(2) do
@sock.read(1).nil?
end
rescue Timeout::Error
return false
end
def write_reply( handle, err=0, opts={} )
def write_reply(handle, err = 0, opts = {})
if opts[:magic] == :wrong
write_rand( @sock, 4 )
write_rand(@sock, 4)
else
@sock.write( ::FlexNBD::REPLY_MAGIC )
@sock.write(::FlexNBD::REPLY_MAGIC)
end
@sock.write( [err].pack("N") )
@sock.write( handle )
@sock.write([err].pack('N'))
@sock.write(handle)
end
def close
@sock.close
end
def read_data( len )
@sock.read( len )
def read_data(len)
@sock.read(len)
end
def write_data( len )
@sock.write( len )
def write_data(len)
@sock.write(len)
end
def self.parse_be64(str)
raise "String is the wrong length: 8 bytes expected (#{str.length} received)" unless
str.length == 8
top, bottom = str.unpack("NN")
top, bottom = str.unpack('NN')
(top << 32) + bottom
end
def receive_mirror( opts = {} )
write_hello()
def receive_mirror(opts = {})
write_hello
loop do
req = read_request
case req[:type]
when 1
read_data( req[:len] )
write_reply( req[:handle] )
when 65536
write_reply( req[:handle], opts[:err] == :entrust ? 1 : 0 )
read_data(req[:len])
write_reply(req[:handle])
when 65_536
write_reply(req[:handle], opts[:err] == :entrust ? 1 : 0)
break
else
raise "Unexpected request: #{req.inspect}"
@@ -129,16 +117,13 @@ module FlexNBD
raise "Not a disconnect: #{req.inspect}"
end
end
end # class Client
def initialize( addr, port )
@sock = TCPServer.new( addr, port )
def initialize(addr, port)
@sock = TCPServer.new(addr, port)
end
def accept( err_msg = "Timed out waiting for a connection", timeout = 5)
def accept(err_msg = 'Timed out waiting for a connection', timeout = 5)
client_sock = nil
begin
@@ -146,21 +131,16 @@ module FlexNBD
client_sock = @sock.accept
end
rescue Timeout::Error
raise Timeout::Error.new(err_msg)
raise Timeout::Error, err_msg
end
client_sock
Client.new( client_sock )
Client.new(client_sock)
end
def close
@sock.close
end
end # module FakeDest
end # module FlexNBD

163
tests/acceptance/flexnbd/fake_source.rb
View File

@@ -1,112 +1,117 @@
# encoding: utf-8
require 'socket'
require "timeout"
require 'timeout'
require 'flexnbd/constants'
module FlexNBD
class FakeSource
def initialize( addr, port, err_msg, source_addr=nil, source_port=0 )
timing_out( 2, err_msg ) do
def initialize(addr, port, err_msg, source_addr = nil, source_port = 0)
timing_out(2, err_msg) do
begin
@sock = if source_addr
TCPSocket.new( addr, port, source_addr, source_port )
TCPSocket.new(addr, port, source_addr, source_port)
else
TCPSocket.new( addr, port )
TCPSocket.new(addr, port)
end
rescue Errno::ECONNREFUSED
$stderr.puts "Connection refused, retrying"
warn 'Connection refused, retrying'
sleep(0.2)
retry
end
end
end
def close
@sock.close
end
def read_hello
timing_out(::FlexNBD::MS_HELLO_TIME_SECS,
'Timed out waiting for hello.') do
raise 'No hello.' unless (hello = @sock.read(152)) &&
hello.length == 152
def read_hello()
timing_out( ::FlexNBD::MS_HELLO_TIME_SECS,
"Timed out waiting for hello." ) do
fail "No hello." unless (hello = @sock.read( 152 )) &&
hello.length==152
passwd_s = hello[0..7]
magic = hello[8..15].unpack('Q>').first
size = hello[16..23].unpack('Q>').first
flags = hello[24..27].unpack('L>').first
reserved = hello[28..-1]
magic_s = hello[0..7]
ignore_s= hello[8..15]
size_s = hello[16..23]
size_h, size_l = size_s.unpack("NN")
size = (size_h << 32) + size_l
return { :magic => magic_s, :size => size }
return { passwd: passwd_s, magic: magic, size: size, flags: flags, reserved: reserved }
end
end
def send_request(type, handle = 'myhandle', from = 0, len = 0, magic = REQUEST_MAGIC, flags = 0)
raise 'Bad handle' unless handle.length == 8
def send_request( type, handle="myhandle", from=0, len=0, magic=REQUEST_MAGIC )
fail "Bad handle" unless handle.length == 8
@sock.write( magic )
@sock.write( [type].pack( 'N' ) )
@sock.write( handle )
@sock.write( [n64( from )].pack( 'q' ) )
@sock.write( [len].pack( 'N' ) )
@sock.write(magic)
@sock.write([flags].pack('n'))
@sock.write([type].pack('n'))
@sock.write(handle)
@sock.write([n64(from)].pack('q'))
@sock.write([len].pack('N'))
end
def write_write_request( from, len, handle="myhandle" )
send_request( 1, handle, from, len )
def write_write_request(from, len, handle = 'myhandle')
send_request(1, handle, from, len)
end
def write_entrust_request( handle="myhandle" )
send_request( 65536, handle )
def write_write_request_with_fua(from, len, handle = 'myhandle')
send_request(1, handle, from, len, REQUEST_MAGIC, 1)
end
def write_disconnect_request( handle="myhandle" )
send_request( 2, handle )
def write_flush_request(handle = 'myhandle')
send_request(3, handle, 0, 0)
end
def write_read_request( from, len, handle="myhandle" )
send_request( 0, "myhandle", from, len )
def write_entrust_request(handle = 'myhandle')
send_request(65_536, handle)
end
def write_data( data )
@sock.write( data )
def write_disconnect_request(handle = 'myhandle')
send_request(2, handle)
end
def write_read_request(from, len, _handle = 'myhandle')
send_request(0, 'myhandle', from, len)
end
def write_data(data)
@sock.write(data)
end
# Handy utility
def read( from, len )
timing_out( 2, "Timed out reading" ) do
send_request( 0, "myhandle", from, len )
read_raw( len )
def read(from, len)
timing_out(2, 'Timed out reading') do
send_request(0, 'myhandle', from, len)
read_raw(len)
end
end
def read_raw( len )
@sock.read( len )
def read_raw(len)
@sock.read(len)
end
def send_mirror
read_hello()
write( 0, "12345678" )
read_response()
write_disconnect_request()
close()
read_hello
write(0, '12345678')
read_response
write_disconnect_request
close
end
def write( from, data )
write_write_request( from, data.length )
write_data( data )
def write(from, data)
write_write_request(from, data.length)
write_data(data)
end
def write_with_fua(from, data)
write_write_request_with_fua(from, data.length)
write_data(data)
end
def flush
write_flush_request
end
def read_response
magic = @sock.read(4)
@@ -114,29 +119,25 @@ module FlexNBD
handle = @sock.read(8)
{
:magic => magic,
:error => error_s.unpack("N").first,
:handle => handle
magic: magic,
error: error_s.unpack('N').first,
handle: handle
}
end
def disconnected?
result = nil
Timeout.timeout( 2 ) { result = ( @sock.read(1) == nil ) }
Timeout.timeout(2) { result = @sock.read(1).nil? }
result
end
def timing_out( time, msg )
begin
Timeout.timeout( time ) do
yield
end
rescue Timeout::Error
$stderr.puts msg
exit 1
def timing_out(time, msg)
Timeout.timeout(time) do
yield
end
rescue Timeout::Error
warn msg
exit 1
end
private
@@ -146,15 +147,13 @@ module FlexNBD
# )
def n64(b)
((b & 0xff00000000000000) >> 56) |
((b & 0x00ff000000000000) >> 40) |
((b & 0x0000ff0000000000) >> 24) |
((b & 0x000000ff00000000) >> 8) |
((b & 0x00000000ff000000) << 8) |
((b & 0x0000000000ff0000) << 24) |
((b & 0x000000000000ff00) << 40) |
((b & 0x00000000000000ff) << 56)
((b & 0x00ff000000000000) >> 40) |
((b & 0x0000ff0000000000) >> 24) |
((b & 0x000000ff00000000) >> 8) |
((b & 0x00000000ff000000) << 8) |
((b & 0x0000000000ff0000) << 24) |
((b & 0x000000000000ff00) << 40) |
((b & 0x00000000000000ff) << 56)
end
end # class FakeSource
end # module FlexNBD

13
tests/acceptance/ld_preloads/Makefile Normal file
View File

@@ -0,0 +1,13 @@
SRC := $(wildcard *.c)
OBJS := $(SRC:%.c=%.o)
all: $(OBJS)
clean:
$(RM) $(OBJS)
%.o: %.c
gcc -shared -fPIC -ldl -o $@ $<
.PHONY: all clean

33
tests/acceptance/ld_preloads/msync_catcher.c Normal file
View File

@@ -0,0 +1,33 @@
#define _GNU_SOURCE
#include <dlfcn.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/mman.h>
typedef int (*real_msync_t)(void *addr, size_t length, int flags);
int real_msync(void *addr, size_t length, int flags) {
return ((real_msync_t)dlsym(RTLD_NEXT, "msync"))(addr, length, flags);
}
/*
* Noddy LD_PRELOAD wrapper to catch msync calls, and log them to a file.
*/
int msync(void *addr, size_t length, int flags) {
FILE *fd;
char *fn;
int retval;
retval = real_msync(addr, length, flags);
fn = getenv("MSYNC_CATCHER_OUTPUT");
if ( fn != NULL ) {
fd = fopen(fn,"a");
fprintf(fd,"msync:%d:%i:%i:%i\n", addr, length, flags, retval);
fclose(fd);
}
return retval;
}

2
tests/acceptance/nbd_scenarios Normal file → Executable file
View File

@@ -1,6 +1,6 @@
#!/usr/bin/ruby
test_files = Dir[File.dirname( __FILE__ ) + "/test*.rb"]
test_files = Dir[File.dirname(__FILE__) + '/test*.rb']
for filename in test_files
require filename
end

70
tests/acceptance/proxy_tests.rb
View File

@@ -1,4 +1,4 @@
# encoding: utf-8
require 'flexnbd/fake_source'
require 'flexnbd/fake_dest'
@@ -7,20 +7,23 @@ module ProxyTests
"\xFF".b
end
def with_proxied_client( override_size = nil )
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 'NBDMAGIC', result[:passwd]
assert_equal override_size || @env.file1.size, result[:size]
yield client
ensure
client.close rescue nil
begin
client.close
rescue StandardError
nil
end
end
end
@@ -32,11 +35,11 @@ module ProxyTests
with_proxied_client do |client|
(0..3).each do |n|
offset = n * 4096
client.write_read_request(offset, 4096, "myhandle")
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 'myhandle', rsp[:handle]
assert_equal 0, rsp[:error]
orig_data = @env.file1.read(offset, 4096)
@@ -45,8 +48,8 @@ module ProxyTests
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}" )
assert_equal(orig_data, data,
"Returned data does not match on request #{n + 1}")
end
end
end
@@ -59,12 +62,12 @@ module ProxyTests
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal "myhandle", rsp[:handle]
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})" )
assert_equal((b * 4096), data, "Data not written correctly (offset is #{n})")
end
end
end
@@ -78,7 +81,7 @@ module ProxyTests
sc = server.accept # just tell the supervisor we're up
sc.write_hello
[ server, sc ]
[server, sc]
end
end
@@ -89,7 +92,7 @@ module ProxyTests
server, sc1 = maker.value
# Send the read request to the proxy
client.write_read_request( 0, 4096 )
client.write_read_request(0, 4096)
# ensure we're given the read request
req1 = sc1.read_request
@@ -110,22 +113,21 @@ module ProxyTests
assert_equal req1, req2
# The reply should be proxied back to the client.
sc2.write_reply( req2[:handle] )
sc2.write_data( b * 4096 )
sc2.write_reply(req2[:handle])
sc2.write_data(b * 4096)
# Check it to make sure it's correct
rsp = timeout(15) { client.read_response }
rsp = Timeout.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" )
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
@@ -135,7 +137,7 @@ module ProxyTests
server, sc1 = maker.value
# Send the read request to the proxy
client.write( 0, ( b * 4096 ) )
client.write(0, (b * 4096))
# ensure we're given the read request
req1 = sc1.read_request
@@ -143,8 +145,8 @@ module ProxyTests
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" )
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
@@ -156,14 +158,14 @@ module ProxyTests
# And once reconnected, it should resend an identical request.
req2 = sc2.read_request
assert_equal req1, req2
data2 = sc2.read_data( 4096 )
data2 = sc2.read_data(4096)
assert_equal data1, data2
# The reply should be proxied back to the client.
sc2.write_reply( req2[:handle] )
sc2.write_reply(req2[:handle])
# Check it to make sure it's correct
rsp = timeout(15) { client.read_response }
rsp = Timeout.timeout(15) { client.read_response }
assert_equal ::FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
assert_equal req1[:handle], rsp[:handle]
@@ -174,21 +176,21 @@ module ProxyTests
end
def test_only_one_client_can_connect_to_proxy_at_a_time
with_proxied_client do |client|
with_proxied_client do |_client|
c2 = nil
assert_raises(Timeout::Error) do
timeout(1) do
Timeout.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
if c2
begin
c2.close
rescue StandardError
nil
end
end
end
end
end

61
tests/acceptance/test_dest_error_handling.rb
View File

@@ -1,13 +1,10 @@
# encoding: utf-8
require 'test/unit'
require 'environment'
class TestDestErrorHandling < Test::Unit::TestCase
class TestDestErrorHandling < Test::Unit::TestCase
def setup
@env = Environment.new
@env.writefile1( "0" * 4 )
@env.writefile1('0' * 4)
@env.listen1
end
@@ -15,89 +12,77 @@ class TestDestErrorHandling < Test::Unit::TestCase
@env.cleanup
end
def test_hello_blocked_by_disconnect_causes_error_not_fatal
run_fake( "source/close_after_connect" )
run_fake('source/close_after_connect')
assert_no_control
end
=begin
# This is disabled while CLIENT_MAX_WAIT_SECS is removed
def test_hello_goes_astray_causes_timeout_error
run_fake( "source/hang_after_hello" )
assert_no_control
end
=end
# # This is disabled while CLIENT_MAX_WAIT_SECS is removed
# def test_hello_goes_astray_causes_timeout_error
# run_fake( "source/hang_after_hello" )
# assert_no_control
# end
def test_sigterm_has_bad_exit_status
@env.nbd1.can_die(1)
run_fake( "source/sigterm_after_hello" )
run_fake('source/sigterm_after_hello')
end
def test_disconnect_after_hello_causes_error_not_fatal
run_fake( "source/close_after_hello" )
run_fake('source/close_after_hello')
assert_no_control
end
def test_partial_read_causes_error
run_fake( "source/close_mid_read" )
run_fake('source/close_mid_read')
end
def test_double_connect_during_hello
run_fake( "source/connect_during_hello" )
run_fake('source/connect_during_hello')
end
def test_acl_rejection
@env.acl1("127.0.0.1")
run_fake( "source/connect_from_banned_ip")
@env.acl1('127.0.0.1')
run_fake('source/connect_from_banned_ip')
end
def test_bad_write
run_fake( "source/write_out_of_range" )
run_fake('source/write_out_of_range')
end
def test_disconnect_before_write_data_causes_error
run_fake( "source/close_after_write" )
run_fake('source/close_after_write')
end
def test_disconnect_before_write_reply_causes_error
# Note that this is an odd case: writing the reply doesn't fail.
# The test passes because the next attempt by flexnbd to read a
# request returns EOF.
run_fake( "source/close_after_write_data" )
run_fake('source/close_after_write_data')
end
def test_straight_migration
@env.nbd1.can_die(0)
run_fake( "source/successful_transfer" )
run_fake('source/successful_transfer')
end
private
def run_fake( name )
@env.run_fake( name, @env.ip, @env.port1 )
def run_fake(name)
@env.run_fake(name, @env.ip, @env.port1)
assert @env.fake_reports_success, "#{name} failed."
end
def status
stat, _ = @env.status1
stat, = @env.status1
stat
end
def assert_no_control
assert !status['has_control'], "Thought it had control"
assert !status['has_control'], 'Thought it had control'
end
def assert_control
assert status['has_control'], "Didn't think it had control"
end
end # class TestDestErrorHandling

100
tests/acceptance/test_happy_path.rb
View File

@@ -1,5 +1,3 @@
# encoding: utf-8
require 'test/unit'
require 'environment'
require 'flexnbd/constants'
@@ -19,20 +17,18 @@ class TestHappyPath < Test::Unit::TestCase
@env.cleanup
end
def test_read1
@env.writefile1("f"*64)
@env.writefile1('f' * 64)
@env.serve1
[0, 12, 63].each do |num|
assert_equal(
bin( @env.nbd1.read(num*@env.blocksize, @env.blocksize) ),
bin( @env.file1.read(num*@env.blocksize, @env.blocksize) )
bin(@env.nbd1.read(num * @env.blocksize, @env.blocksize)),
bin(@env.file1.read(num * @env.blocksize, @env.blocksize))
)
end
[124, 1200, 10028, 25488].each do |num|
[124, 1200, 10_028, 25_488].each do |num|
assert_equal(bin(@env.nbd1.read(num, 4)), bin(@env.file1.read(num, 4)))
end
end
@@ -40,14 +36,14 @@ class TestHappyPath < Test::Unit::TestCase
# Check that we're not
#
def test_writeread1
@env.writefile1("0"*64)
@env.writefile1('0' * 64)
@env.serve1
[0, 12, 63].each do |num|
data = "X"*@env.blocksize
@env.nbd1.write(num*@env.blocksize, data)
assert_equal(data, @env.file1.read(num*@env.blocksize, data.size))
assert_equal(data, @env.nbd1.read(num*@env.blocksize, data.size))
data = 'X' * @env.blocksize
@env.nbd1.write(num * @env.blocksize, data)
assert_equal(data, @env.file1.read(num * @env.blocksize, data.size))
assert_equal(data, @env.nbd1.read(num * @env.blocksize, data.size))
end
end
@@ -55,115 +51,105 @@ class TestHappyPath < Test::Unit::TestCase
# up.
#
def test_writeread2
@env.writefile1("0"*1024)
@env.writefile1('0' * 1024)
@env.serve1
d0 = "\0"*@env.blocksize
d1 = "X"*@env.blocksize
d0 = "\0" * @env.blocksize
d1 = 'X' * @env.blocksize
(0..63).each do |num|
@env.nbd1.write(num*@env.blocksize*2, d1)
@env.nbd1.write(num * @env.blocksize * 2, d1)
end
(0..63).each do |num|
assert_equal(d0, @env.nbd1.read(((2*num)+1)*@env.blocksize, d0.size))
assert_equal(d0, @env.nbd1.read(((2 * num) + 1) * @env.blocksize, d0.size))
end
end
def setup_to_mirror
@env.writefile1( "f"*4 )
@env.writefile1('f' * 4)
@env.serve1
@env.writefile2( "0"*4 )
@env.writefile2('0' * 4)
@env.listen2
end
def test_mirror
@env.nbd1.can_die
@env.nbd2.can_die(0)
setup_to_mirror()
setup_to_mirror
stdout, stderr = @env.mirror12
@env.nbd1.join
@env.nbd2.join
assert( File.file?( @env.filename1 ),
"The source file was incorrectly deleted")
assert_equal(@env.file1.read_original( 0, @env.blocksize ),
@env.file2.read( 0, @env.blocksize ) )
assert(File.file?(@env.filename1),
'The source file was incorrectly deleted')
assert_equal(@env.file1.read_original(0, @env.blocksize),
@env.file2.read(0, @env.blocksize))
end
def test_mirror_unlink
@env.nbd1.can_die(0)
@env.nbd2.can_die(0)
setup_to_mirror()
setup_to_mirror
assert File.file?( @env.filename1 )
assert File.file?(@env.filename1)
stdout, stderr = @env.mirror12_unlink
assert_no_match( /unrecognized/, stderr )
assert_no_match(/unrecognized/, stderr)
Timeout.timeout(10) { @env.nbd1.join }
Timeout.timeout(10) do @env.nbd1.join end
assert !File.file?( @env.filename1 )
assert !File.file?(@env.filename1)
end
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 )
#
# 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.truncate1 '4G'
@env.serve1
@env.nbd1.write( 2**31+2**29, "12345678" )
@env.nbd1.write(2**31 + 2**29, '12345678')
sleep(1)
assert_equal "12345678", @env.nbd1.read( 2**31+2**29, 8 )
assert_equal '12345678', @env.nbd1.read(2**31 + 2**29, 8)
end
def test_set_acl
# Just check that we get sane feedback here
@env.writefile1( "f"*4 )
@env.writefile1('f' * 4)
@env.serve1
_,stderr = @env.acl1("127.0.0.1")
assert_no_match( /^(F|E):/, stderr )
_, stderr = @env.acl1('127.0.0.1')
assert_no_match(/^(F|E):/, stderr)
end
def test_write_more_than_one_run
one_mb = 2**20
data = "\0" * 256 * one_mb
File.open(@env.filename1, "wb") do |f| f.write( "1" * 256 * one_mb ) end
File.open(@env.filename1, 'wb') { |f| f.write('1' * 256 * one_mb) }
@env.serve1
sleep 5
@env.write1( data )
@env.write1(data)
@env.nbd1.can_die(0)
@env.nbd1.kill
i = 0
File.open(@env.filename1, "rb") do |f|
while mb = f.read( one_mb )
unless "\0"*one_mb == mb
msg = "Read non-zeros after offset %x:\n"%(i * one_mb)
File.open(@env.filename1, 'rb') do |f|
while mb = f.read(one_mb)
unless "\0" * one_mb == mb
msg = format("Read non-zeros after offset %x:\n", (i * one_mb))
msg += `hexdump #{@env.filename1} | head -n5`
fail msg
raise msg
end
i += 1
end
end
end
end

5
tests/acceptance/test_prefetch_proxy_mode.rb
View File

@@ -2,7 +2,6 @@ require 'test/unit'
require 'environment'
require 'proxy_tests'
class TestPrefetchProxyMode < Test::Unit::TestCase
include ProxyTests
@@ -10,7 +9,7 @@ class TestPrefetchProxyMode < Test::Unit::TestCase
super
@env = Environment.new
@env.prefetch_proxy!
@env.writefile1( "f" * 16 )
@env.writefile1('f' * 16)
end
def teardown
@@ -18,5 +17,3 @@ class TestPrefetchProxyMode < Test::Unit::TestCase
super
end
end

4
tests/acceptance/test_proxy_mode.rb
View File

@@ -2,14 +2,13 @@ require 'test/unit'
require 'environment'
require 'proxy_tests'
class TestProxyMode < Test::Unit::TestCase
include ProxyTests
def setup
super
@env = Environment.new
@env.writefile1( "f" * 16 )
@env.writefile1('f' * 16)
end
def teardown
@@ -17,4 +16,3 @@ class TestProxyMode < Test::Unit::TestCase
super
end
end

116
tests/acceptance/test_serve_mode.rb
View File

@@ -1,86 +1,122 @@
require 'test/unit'
require 'environment'
require 'flexnbd/fake_source'
require 'tempfile'
class TestServeMode < Test::Unit::TestCase
def setup
super
@b = "\xFF".b
@env = Environment.new
@env.writefile1( "0" )
@env.serve1
end
def teardown
@env.cleanup
teardown_msync_catcher
super
end
def connect_to_server
client = FlexNBD::FakeSource.new(@env.ip, @env.port1, "Connecting to server failed")
@env.writefile1('0')
@env.serve1
client = FlexNBD::FakeSource.new(@env.ip, @env.port1, 'Connecting to server failed')
begin
result = client.read_hello
assert_equal "NBDMAGIC", result[:magic]
assert_equal 'NBDMAGIC', result[:passwd]
assert_equal 0x00420281861253, result[:magic]
assert_equal @env.file1.size, result[:size]
# See src/common/nbdtypes.h for the various flags. At the moment we
# support HAS_FLAGS (1), SEND_FLUSH (4), SEND_FUA (8)
assert_equal (1 | 4 | 8), result[:flags]
assert_equal "\x0" * 124, result[:reserved]
yield client
ensure
client.close rescue nil
begin
client.close
rescue StandardError
nil
end
end
end
def setup_msync_catcher
`make -C ld_preloads/ msync_catcher.o`
omit 'LD_PRELOAD library not found' unless
File.exist?('ld_preloads/msync_catcher.o')
@msync_catcher = Tempfile.new('msync')
ENV['MSYNC_CATCHER_OUTPUT'] = @msync_catcher.path
@ld_preload_orig = ENV['LD_PRELOAD']
ENV['LD_PRELOAD'] = 'ld_preloads/msync_catcher.o'
end
def parse_msync_output
op = []
until @msync_catcher.eof?
op << @msync_catcher.readline.chomp.split(':').map do |e|
e =~ /^\d+$/ ? e.to_i : e
end
end
op
end
def teardown_msync_catcher
@msync_catcher.close if @msync_catcher
ENV.delete 'MSYNC_CATCHER_OUTPUT'
ENV['LD_PRELOAD'] = @ld_preload_orig
end
def test_bad_request_magic_receives_error_response
connect_to_server do |client|
# replace REQUEST_MAGIC with all 0s to make it look bad
client.send_request( 0, "myhandle", 0, 0, "\x00\x00\x00\x00" )
client.send_request(0, 'myhandle', 0, 0, "\x00\x00\x00\x00")
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal "myhandle", rsp[:handle]
assert_equal 'myhandle', rsp[:handle]
assert rsp[:error] != 0, "Server sent success reply back: #{rsp[:error]}"
# The client should be disconnected now
assert client.disconnected?, "Server not disconnected"
assert client.disconnected?, 'Server not disconnected'
end
end
def test_long_write_on_top_of_short_write_is_respected
connect_to_server do |client|
# Start with a file of all-zeroes.
client.write( 0, "\x00" * @env.file1.size )
client.write(0, "\x00" * @env.file1.size)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
client.write( 0, @b )
client.write(0, @b)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
client.write( 0, @b * 2 )
client.write(0, @b * 2)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
end
assert_equal @b * 2, @env.file1.read( 0, 2 )
assert_equal @b * 2, @env.file1.read(0, 2)
end
def test_read_request_out_of_bounds_receives_error_response
connect_to_server do |client|
client.write_read_request( @env.file1.size, 4096 )
client.write_read_request(@env.file1.size, 4096)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal "myhandle", rsp[:handle]
assert_equal 'myhandle', rsp[:handle]
assert rsp[:error] != 0, "Server sent success reply back: #{rsp[:error]}"
# Ensure we're not disconnected by sending a request. We don't care about
# whether the reply is good or not, here.
client.write_read_request( 0, 4096 )
client.write_read_request(0, 4096)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
end
@@ -88,23 +124,55 @@ class TestServeMode < Test::Unit::TestCase
def test_write_request_out_of_bounds_receives_error_response
connect_to_server do |client|
client.write( @env.file1.size, "\x00" * 4096 )
client.write(@env.file1.size, "\x00" * 4096)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal "myhandle", rsp[:handle]
assert_equal 'myhandle', rsp[:handle]
assert rsp[:error] != 0, "Server sent success reply back: #{rsp[:error]}"
# Ensure we're not disconnected by sending a request. We don't care about
# whether the reply is good or not, here.
client.write( 0, "\x00" * @env.file1.size )
client.write(0, "\x00" * @env.file1.size)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
end
end
def test_flush_is_accepted
setup_msync_catcher
connect_to_server do |client|
client.flush
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
end
op = parse_msync_output
assert_equal 1, op.count, 'Only one msync expected'
assert_equal @env.blocksize, op.first[2], 'msync length wrong'
assert_equal 6, op.first[3], 'msync called with incorrect flags'
end
def test_write_with_fua_is_accepted
setup_msync_catcher
page_size = Integer(`getconf PAGESIZE`)
@env.blocksize = page_size * 10
connect_to_server do |client|
# Write somewhere in the third page
pos = page_size * 3 + 100
client.write_with_fua(pos, "\x00" * 33)
rsp = client.read_response
assert_equal FlexNBD::REPLY_MAGIC, rsp[:magic]
assert_equal 0, rsp[:error]
end
op = parse_msync_output
assert_equal 1, op.count, 'Only one msync expected'
# Should be 100 + 33, as we've started writing 100 bytes into a page, for
# 33 bytes
assert_equal 133, op.first[2], 'msync length wrong'
assert_equal 6, op.first[3], 'msync called with incorrect flags'
end
end

77
tests/acceptance/test_source_error_handling.rb
View File

@@ -1,126 +1,105 @@
# encoding: utf-8
require 'test/unit'
require 'environment'
class TestSourceErrorHandling < Test::Unit::TestCase
def setup
@old_env = ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS']
ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS'] = "4.0"
ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS'] = '4.0'
@env = Environment.new
@env.writefile1( "f" * 4 )
@env.writefile1('f' * 4)
@env.serve1
end
def teardown
@env.nbd1.can_die(0)
@env.cleanup
ENV['FLEXNBD_MS_REQUEST_LIMIT_SECS'] = @old_env
end
def expect_term_during_migration
@env.nbd1.can_die(1,9)
@env.nbd1.can_die(1, 9)
end
def test_failure_to_connect_reported_in_mirror_cmd_response
stdout, stderr = @env.mirror12_unchecked
expect_term_during_migration
assert_match( /failed to connect/, stderr )
assert_match(/failed to connect/, stderr)
end
def test_sigterm_after_hello_quits_with_status_of_1
expect_term_during_migration
run_fake( "dest/sigterm_after_hello" )
run_fake('dest/sigterm_after_hello')
end
def test_destination_hangs_after_connect_reports_error_at_source
run_fake( "dest/hang_after_connect",
:err => /Remote server failed to respond/ )
run_fake('dest/hang_after_connect',
err: /Remote server failed to respond/)
end
def test_destination_rejects_connection_reports_error_at_source
run_fake( "dest/reject_acl",
:err => /Mirror was rejected/ )
run_fake('dest/reject_acl',
err: /Mirror was rejected/)
end
def test_wrong_size_causes_disconnect
run_fake( "dest/hello_wrong_size",
:err => /Remote size does not match local size/ )
run_fake('dest/hello_wrong_size',
err: /Remote size does not match local size/)
end
def test_wrong_magic_causes_disconnect
expect_term_during_migration
run_fake( "dest/hello_wrong_magic",
:err => /Mirror was rejected/ )
run_fake('dest/hello_wrong_magic',
err: /Mirror was rejected/)
end
def test_disconnect_after_hello_causes_retry
expect_term_during_migration
run_fake( "dest/close_after_hello",
:out => /Mirror started/ )
run_fake('dest/close_after_hello',
out: /Mirror started/)
end
def test_write_times_out_causes_retry
expect_term_during_migration
run_fake( "dest/hang_after_write" )
run_fake('dest/hang_after_write')
end
def test_rejected_write_causes_retry
expect_term_during_migration
run_fake( "dest/error_on_write" )
run_fake('dest/error_on_write')
end
def test_disconnect_before_write_reply_causes_retry
expect_term_during_migration
run_fake( "dest/close_after_write" )
run_fake('dest/close_after_write')
end
def test_bad_write_reply_causes_retry
expect_term_during_migration
run_fake( "dest/write_wrong_magic" )
run_fake('dest/write_wrong_magic')
end
def test_pre_entrust_disconnect_causes_retry
expect_term_during_migration
run_fake( "dest/close_after_writes" )
run_fake('dest/close_after_writes')
end
def test_cancel_migration
run_fake( "dest/break_after_hello" )
run_fake('dest/break_after_hello')
end
private
def run_fake(name, opts = {})
@env.run_fake( name, @env.ip, @env.port2, @env.nbd1.ctrl )
@env.run_fake(name, @env.ip, @env.port2, @env.nbd1.ctrl)
stdout, stderr = @env.mirror12_unchecked
assert_success
assert_match( opts[:err], stderr ) if opts[:err]
assert_match( opts[:out], stdout ) if opts[:out]
return stdout, stderr
assert_match(opts[:err], stderr) if opts[:err]
assert_match(opts[:out], stdout) if opts[:out]
[stdout, stderr]
end
def assert_success( msg=nil )
assert @env.fake_reports_success, msg || "Fake failed"
def assert_success(msg = nil)
assert @env.fake_reports_success, msg || 'Fake failed'
end
end # class TestSourceErrorHandling

121
tests/acceptance/test_write_during_migration.rb Executable file → Normal file
View File

@@ -9,102 +9,98 @@ require 'tmpdir'
Thread.abort_on_exception = true
class TestWriteDuringMigration < Test::Unit::TestCase
def setup
@flexnbd = File.expand_path("../../build/flexnbd")
@flexnbd = File.expand_path('../../build/flexnbd')
raise "No binary!" unless File.executable?( @flexnbd )
raise 'No binary!' unless File.executable?(@flexnbd)
@size = 20*1024*1024 # 20MB
@write_data = "foo!" * 2048 # 8K write
@size = 20 * 1024 * 1024 # 20MB
@write_data = 'foo!' * 2048 # 8K write
@source_port = 9990
@dest_port = 9991
@source_sock = "src.sock"
@dest_sock = "dst.sock"
@source_file = "src.file"
@dest_file = "dst.file"
@source_sock = 'src.sock'
@dest_sock = 'dst.sock'
@source_file = 'src.file'
@dest_file = 'dst.file'
end
def teardown
[@dst_proc, @src_proc].each do |pid|
if pid
Process.kill( "KILL", pid ) rescue nil
next unless pid
begin
Process.kill('KILL', pid)
rescue StandardError
nil
end
end
end
def debug_arg
ENV['DEBUG'] ? "--verbose" : ""
ENV['DEBUG'] ? '--verbose' : ''
end
def launch_servers
@dst_proc = fork() {
@dst_proc = fork do
cmd = "#{@flexnbd} listen -l 127.0.0.1 -p #{@dest_port} -f #{@dest_file} -s #{@dest_sock} #{debug_arg}"
exec cmd
}
end
@src_proc = fork() {
@src_proc = fork do
cmd = "#{@flexnbd} serve -l 127.0.0.1 -p #{@source_port} -f #{@source_file} -s #{@source_sock} #{debug_arg}"
exec cmd
}
end
begin
awaiting = nil
Timeout.timeout(10) do
awaiting = :source
sleep 0.1 while !File.exists?( @source_sock )
sleep 0.1 until File.exist?(@source_sock)
awaiting = :dest
sleep 0.1 while !File.exists?( @dest_sock )
sleep 0.1 until File.exist?(@dest_sock)
end
rescue Timeout::Error
case awaiting
when :source
fail "Couldn't get a source socket."
raise "Couldn't get a source socket."
when :dest
fail "Couldn't get a destination socket."
raise "Couldn't get a destination socket."
else
fail "Something went wrong I don't understand."
raise "Something went wrong I don't understand."
end
end
end
def make_files()
def make_files
FileUtils.touch(@source_file)
File.truncate(@source_file, @size)
FileUtils.touch(@dest_file)
File.truncate(@dest_file, @size)
File.open(@source_file, "wb"){|f| f.write "a"*@size }
File.open(@source_file, 'wb') { |f| f.write 'a' * @size }
end
def start_mirror
UNIXSocket.open(@source_sock) {|sock|
sock.write(["mirror", "127.0.0.1", @dest_port.to_s, "exit"].join("\x0A") + "\x0A\x0A")
UNIXSocket.open(@source_sock) do |sock|
sock.write(['mirror', '127.0.0.1', @dest_port.to_s, 'exit'].join("\x0A") + "\x0A\x0A")
sock.flush
rsp = sock.readline
}
end
end
def wait_for_quit()
Timeout.timeout( 10 ) do
start_time = Time.now
dst_result = Process::waitpid2(@dst_proc)
src_result = Process::waitpid2(@src_proc)
def wait_for_quit
Timeout.timeout(10) do
start_time = Time.now
dst_result = Process.waitpid2(@dst_proc)
src_result = Process.waitpid2(@src_proc)
end
end
def source_writer
client = FlexNBD::FakeSource.new( "127.0.0.1", @source_port, "Timed out connecting" )
offsets = Range.new(0, (@size - @write_data.size) / 4096 ).to_a
client = FlexNBD::FakeSource.new('127.0.0.1', @source_port, 'Timed out connecting')
offsets = Range.new(0, (@size - @write_data.size) / 4096).to_a
loop do
begin
client.write(offsets[rand(offsets.size)] * 4096, @write_data)
rescue => err
rescue StandardError => err
# We expect a broken write at some point, so ignore it
break
end
@@ -115,32 +111,32 @@ class TestWriteDuringMigration < Test::Unit::TestCase
# puts `md5sum #{@source_file} #{@dest_file}`
# Ensure each block matches
File.open(@source_file, "r") do |source|
File.open(@dest_file, "r") do |dest|
0.upto( @size / 4096 ) do |block_num|
s_data = source.read( 4096 )
d_data = dest.read( 4096 )
File.open(@source_file, 'r') do |source|
File.open(@dest_file, 'r') do |dest|
0.upto(@size / 4096) do |block_num|
s_data = source.read(4096)
d_data = dest.read(4096)
assert s_data == d_data, "Block #{block_num} mismatch!"
source.seek( 4096, IO::SEEK_CUR )
dest.seek( 4096, IO::SEEK_CUR )
source.seek(4096, IO::SEEK_CUR)
dest.seek(4096, IO::SEEK_CUR)
end
end
end
end
def test_write_during_migration
Dir.mktmpdir() do |tmpdir|
Dir.chdir( tmpdir ) do
make_files()
Dir.mktmpdir do |tmpdir|
Dir.chdir(tmpdir) do
make_files
launch_servers()
launch_servers
src_writer = Thread.new { source_writer }
start_mirror()
wait_for_quit()
start_mirror
wait_for_quit
src_writer.join
assert_both_sides_identical
end
@@ -148,24 +144,21 @@ class TestWriteDuringMigration < Test::Unit::TestCase
end
def test_many_clients_during_migration
Dir.mktmpdir() do |tmpdir|
Dir.chdir( tmpdir ) do
make_files()
Dir.mktmpdir do |tmpdir|
Dir.chdir(tmpdir) do
make_files
launch_servers()
launch_servers
src_writers_1 = (1..5).collect { Thread.new { source_writer } }
start_mirror()
start_mirror
src_writers_2 = (1..5).collect { Thread.new { source_writer } }
wait_for_quit()
( src_writers_1 + src_writers_2 ).each {|t| t.join }
wait_for_quit
(src_writers_1 + src_writers_2).each(&:join)
assert_both_sides_identical
end
end end
end end
end

27
tests/unit/check_nbdtypes.c
View File

@@ -66,18 +66,35 @@ START_TEST(test_request_magic )
}
END_TEST
START_TEST(test_request_type )
START_TEST(test_request_type)
{
struct nbd_request_raw request_raw;
struct nbd_request request;
request_raw.type = 12345;
request_raw.type = 123;
nbd_r2h_request( &request_raw, &request );
fail_unless( be32toh( 12345 ) == request.type, "Type was not converted." );
fail_unless( be16toh( 123 ) == request.type, "Type was not converted." );
request.type = 67890;
request.type = 234;
nbd_h2r_request( &request, &request_raw );
fail_unless( htobe32( 67890 ) == request_raw.type, "Type was not converted back." );
fail_unless( htobe16( 234 ) == request_raw.type, "Type was not converted back." );
}
END_TEST
START_TEST(test_request_flags)
{
struct nbd_request_raw request_raw;
struct nbd_request request;
request_raw.flags = 123;
nbd_r2h_request( &request_raw, &request );
fail_unless( be16toh( 123 ) == request.flags, "Flags were not converted." );
request.flags = 234;
nbd_h2r_request( &request, &request_raw );
fail_unless( htobe16( 234 ) == request_raw.flags, "Flags were not converted back." );
}
END_TEST