Home | Forums | SDKs | SDKs

using TAP for continuous replication

1 reply [Last post]
Mon, 04/02/2012 - 20:05
nkrishnan
Offline
Joined: 04/02/2012
Groups: None

I am attempting to set up a listener that would get all updates to a couchbase installation. Typically the listener would start up, get all updates since the beginning of time, _and continue to receive updates_ as they get made to the database in the future. The problem I am running into is, after consuming all past changes, the connection times out (default 2.5 seconds) and the program aborts. In couchdb there was a mechanism to set up a db/_continuos connection and have a heartbeat message sent every N seconds to prevent timeout. Can something similar be done here? Its possible for instance that there may be no updates for several hours or days, but we still want the client to remain connected and receive updates when they become available.

This is the listener:

struct libcouchbase_io_opt_st * io;
libcouchbase_t instance;
char const * conf_file = argv[1];
char const * server =
char const * bucket =
io = libcouchbase_create_io_ops( LIBCOUCHBASE_IO_OPS_DEFAULT, NULL, NULL );
if( io == NULL )
exit(-1);
instance = libcouchbase_create( server, NULL, NULL, bucket, io );
if( instance == NULL )
exit(-2);
libcouchbase_set_error_callback(instance, error_callback);
if( libcouchbase_connect( instance ) != LIBCOUCHBASE_SUCCESS ) {
exit(-3);
}
libcouchbase_wait(instance);
libcouchbase_set_tap_mutation_callback( instance, tap_mutation );
libcouchbase_tap_cluster(instance, NULL, NULL, 0 );
libcouchbase_wait(instance);

(the tap_mutation callback just saves the records to disk)

This is the stacktrace:

Program received signal SIGABRT, Aborted.
0x00002aaaadc0d265 in raise () from /lib64/libc.so.6
(gdb) bt
#0 0x00002aaaadc0d265 in raise () from /lib64/libc.so.6
#1 0x00002aaaadc0ed10 in abort () from /lib64/libc.so.6
#2 0x00002aaaace3776a in libcouchbase_purge_single_server (server=0x643e00, stream=0x643e70, cookies=0x643e98, tmo=2500000000, now=1333419708269847000,
error=LIBCOUCHBASE_ETIMEDOUT) at src/server.c:224
#3 0x00002aaaace39721 in libcouchbase_purge_timedout (instance=0x6101f0) at src/timeout.c:94
#4 0x00002aaaace397bb in libcouchbase_timeout_handler (sock=, which=, arg=0x6101f0) at src/timeout.c:41
#5 0x00002aaaac079079 in event_base_loop () from /usr/lib64/libevent-2.0.so.5

this is set off by the last libcouchbase_wait(instance)

the abort seems to happen because libcouchbase_purge_single_server does not handle PROTOCOL_BINARY_CMD_TAP_CONNECT and the default action is abort

Im using:
couchbase-1.8.0

libcouchbase-1.0.2
libvbucket-1.8.0.3

Top
Mon, 04/02/2012 - 23:51
ingenthr
Offline
Joined: 03/16/2010
Groups:

There actually is something similar already. I'd have to look up the specific conditions, but I believe if it's named tap with ACK you'll receive a noop every n seconds, with some default of 600 seconds I think. It's reasonably high.

It's not directly in libcouchbase, but see the series of commits around here: https://github.com/membase/vbucketmigrator/commits/master?page=3

I'd have to read some ep-engine code to verify the behavior I'm describing, but it may be just as easy to wireshark and watch for a noop in an idle state. The basic idea was that the client would be programmed to drop and reconnect if data is not received in ${interval} and the server would be programmed to send a noop over tap in ${interval/2}

Top