WIP: consumer based communication

config-available/redis.pl

@@ -8,3 +8,4 @@
 :- multifile swish_config:config/2.
 swish_config:config(redis, swish).
 swish_config:config(redis_prefix, swish).
+%swish_config:config(redis_consumer, alice).

lib/gitty_driver_files.pl

@@ -75,6 +75,10 @@
 :- use_module(library(option)).
 :- use_module(library(dcg/basics)).
 
+:- use_module(swish_redis).
+:- use_module(library(redis)).
+:- use_module(library(redis_streams)).
+
 /** <module> Gitty plain files driver
 
 This version of the driver uses plain files  to store the gitty data. It
@@ -140,8 +144,7 @@
     option(redis(DB), Options),
     !,
     option(redis_prefix(Prefix), Options, swish),
-    asserta(redis_db(Store, DB, Prefix)),
-    redis_listen(Store).
+    asserta(redis_db(Store, DB, Prefix)).
 gitty_open(_, _).
 
 
@@ -237,24 +240,25 @@
     make_directory(Dir).
 :- endif.
 
-%!  store_object_raw(+Store, +Hash, +Bytes:string) is det.
+%!  store_object_raw(+Store, +Hash, +Bytes:string, -New) is det.
 %
 %   Store an object  from  raw  bytes.   This  is  used  for replicating
 %   objects.
 
-store_object_raw(Store, Hash, _Bytes) :-
+store_object_raw(Store, Hash, _Bytes, false) :-
     pack_object(Hash, _Type, _Size, _Offset, Store, _Pack),
     !.
-store_object_raw(Store, Hash, Bytes) :-
+store_object_raw(Store, Hash, Bytes, New) :-
     gitty_object_file(Store, Hash, Path),
     with_mutex(gitty_file, exists_or_create(Path, Out)),
     (   var(Out)
-    ->  true
+    ->  New = false
     ;   call_cleanup(
             ( set_stream(Out, type(binary)),
               write(Out, Bytes)
             ),
-            close(Out))
+            close(Out)),
+        New = true
     ).
 
 %!  load_object(+Store, +Hash, -Data, -Type, -Size) is det.
@@ -1258,70 +1262,53 @@
 
 %!  redis_replicate_get(+Store, +Hash)
 %
-%   Try to get an object from another SWISH server in the network.
-
-redis_replicate_get(Store, Hash) :-
-    redis_db(Store, DB, Prefix),
-    repl_key(Prefix, Hash, Key),
-    between(0, 3, I),
-    (   redis(DB, get(Key), Data)
-    ->  !,
-        debug(redis(replicate), 'Received object ~p', [Hash]),
-        store_object_raw(Store, Hash, Data)
-    ;   debug(redis(replicate), 'Asking object ~p', [Hash]),
-        publish(Store, gitty(replicate(Hash))),
-        Sleep is min(10, (1<<I)*0.05),
-        sleep(Sleep),
-        fail
-    ).
-
-repl_key(Prefix, Hash, Key) :-
-    atomics_to_string([Prefix,gitty,object,Hash], ":", Key).
+%   Try to get an object from another   SWISH  server in the network. We
+%   implement replication using two streams:
+%
+%     - <prefix>:gitty:replicate
+%     - <prefix>:gitty:discover
+%
+%
 
-publish(Store, Term) :-
-    redis_db(Store, DB, Prefix),
-    redis(DB, publish(Prefix, prolog(Term)), _).
+:- multifile
+    swish_redis:stream/2.
 
-%!  redis_listen(+Store)
-%
-%   Start a redis client to listen for events on store.
+swish_redis:stream('gitty:replicate', [maxlen(1000)]).
+swish_redis:stream('gitty:discover',  [maxlen(1000)]).
 
-redis_listen(Store) :-
-    redis_db(Store, DB, Key),
-    !,
-    catch(thread_create(redis_listen(DB, Key), _,
-                        [ alias(redis) ]),
-          E, print_message(warning, E)).
-redis_listen(_).
-
-redis_listen(DB, Key) :-
-    redis_connect(DB, Connection, []),
-    redis_write(Connection, subscribe(Key)),
-    listen_loop(DB, Key, Connection).
-
-listen_loop(DB, Key, Connection) :-
-    repeat,
-    (   catch(redis_read(Connection, Message), E, true),
-        var(E)
-    ->  catch(redis_dispatch(Message), E,
-              print_message(warning, E)),
-        fail
-    ;   redis_listen(DB, Key)
-    ).
+:- listen(http(pre_server_start(_)),
+          init_replicator).
 
-redis_dispatch(["message", _Channel, Data]) :-
-    debug(redis(subscribe), 'Subscribe: got ~p', [Data]),
-    dispatch(Data).
+init_replicator :-
+    redis_swish_stream('gitty:replicate', ReplKey),
+    listen(redis_consume(ReplKey, Data1, Context1),
+           replicate(Data1, Context1)),
+    redis_swish_stream('gitty:discover', DiscKey),
+    listen(redis_consume(DiscKey, Data2, Context2),
+           discover(Data2, Context2)),
+    message_queue_create(_, [alias(gitty_queue)]).
 
-dispatch(gitty(replicate(Hash))) :-
+discover(Data, _Context) :-
     store(Store, _),
     load_object_raw(Store, Hash, Data),
-    redis_db(Store, DB, Prefix),
-    repl_key(Prefix, Hash, Key),
-    (   redis(DB, exists(Key), 1)
-    ->  true
-    ;   redis(DB, set(Key, Data, ex, 60), _)
-    ).
+    debug(gitty(replicate), 'Sending object ~p', [Hash]),
+    redis_swish_stream('gitty:replicate', ReplKey),
+    xadd(swish, ReplKey, _, _{hash:Hash, data:Data}).
+
+replicate(Data, _Context) :-
+    redis_db(Store, _DB, _Prefix),
+    atom_string(Hash, Data.hash),
+    store_object_raw(Store, Hash, Data.data, _0New),
+    debug(gitty(replicate), 'Received object ~p (new=~p)', [Hash, _0New]),
+    thread_send_message(gitty_queue, Hash).
+
+redis_replicate_get(Store, Hash) :-
+    redis_db(Store, _DB, _Prefix),
+    redis_swish_stream('gitty:discover', DiscKey),
+    xadd(swish, DiscKey, _, _{hash:Hash}),
+    thread_get_message(gitty_queue, Hash,
+                       [ timeout(10)
+                       ]).
 
 
 		 /*******************************

lib/swish_redis.pl

@@ -0,0 +1,118 @@
+/*  Part of SWISH
+
+    Author:        Jan Wielemaker
+    E-mail:        [email protected]
+    WWW:           http://www.swi-prolog.org
+    Copyright (C): 2020, VU University Amsterdam
+			 CWI Amsterdam
+    All rights reserved.
+
+    Redistribution and use in source and binary forms, with or without
+    modification, are permitted provided that the following conditions
+    are met:
+
+    1. Redistributions of source code must retain the above copyright
+       notice, this list of conditions and the following disclaimer.
+
+    2. Redistributions in binary form must reproduce the above copyright
+       notice, this list of conditions and the following disclaimer in
+       the documentation and/or other materials provided with the
+       distribution.
+
+    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+    POSSIBILITY OF SUCH DAMAGE.
+*/
+
+:- module(swish_redis,
+          [ reinit_redis/0,
+            redis_swish_stream/2
+          ]).
+:- use_module(library(redis)).
+:- use_module(library(redis_streams)).
+:- use_module(library(broadcast)).
+:- use_module(library(option)).
+:- use_module(library(socket)).
+
+:- use_module(config).
+
+/** <module> Redis stream connection
+
+Setup a thread to listen to redis connections.
+
+Note that config-available sets up  the   redis  server  using the alias
+`swish`. Streams (redis keys) to  listen   on  are  registered using the
+multifile predicate stream/2.
+*/
+
+:- multifile
+    stream/2.
+
+:- listen(http(pre_server_start(Port)),
+          init_redis(Port)).
+
+:- dynamic port/1.
+
+init_redis(_Port) :-
+    catch(thread_property(redis_listener, id(_)), error(_,_), fail),
+    !.
+init_redis(Port) :-
+    retractall(port(_)),
+    asserta(port(Port)),
+    findall(S, stream(S), Streams),
+    consumer(Port, Consumer),
+    thread_create(xlisten_group(swish, swish, Consumer, Streams,
+                                [ block(1)
+                                ]),
+                  _, [ alias(redis_listener)
+                     ]).
+
+%!  reinit_redis
+%
+%   Stop and start the redis thread. May   be  used to reconfigure it or
+%   restart when crashed.
+
+reinit_redis :-
+    catch(stop_listener, error(_,_), true),
+    port(Port),
+    init_redis(Port).
+
+stop_listener :-
+    thread_signal(redis_listener, redis(stop(false))),
+    thread_join(redis_listener, _).
+
+stream(Key) :-
+    stream(Name, Options),
+    redis_swish_stream(Name, Key),
+    option(max_len(MaxLen), Options, 1000),
+    add_consumer_group(Key),
+    xstream_set(swish, Key, maxlen(MaxLen)).
+
+add_consumer_group(Key) :-
+    catch(redis(swish, xgroup(create, Key, swish, $, mkstream), _),
+          error(redis_error(busygroup,_),_),
+          true).
+
+redis_swish_stream(Name, Key) :-
+    swish_config(redis_prefix, Prefix),
+    atomic_list_concat([Prefix, Name], :, Key).
+
+consumer(_, Consumer) :-
+    swish_config(redis_consumer, Consumer),
+    !.
+consumer(Host:Port, Consumer) :-
+    !,
+    atomic_list_concat([Host,Port], :, Consumer).
+consumer(Port, Consumer) :-
+    gethostname(Host),
+    atomic_list_concat([Host,Port], :, Consumer).
+