Atomic operations on Broker store

Zeek
1

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA256

Hi all,

I am playing around with Broker framework and I am having some trouble
trying to share elements such as set or table between multiple instances
of Bro.

What I am doing follows :

* A master store script creates a table and add it to the store
* A frontend store script retreive the table, add elements to it and
push it back to the store

I got strange results doing that such as only one element I was trying
to add was added (never the same) so I suspected the problem was
concurrency and that my operations were not atomic. Indeed, what happens
it that every call to my function in the slave script do not retreive an
up-to-date table. Retreiving and adding a new element to a table is not
an atomic operation and no function in the API is defined to do it in
one call.

If I do the same test with Broker::add_to_set() with a set instead of a
table it works. And it seems that it is because it is implemented as an
atomic operation in aux/broker/src/store/frontend.cc.

Any idea how to that with a table with existing functions? Or does it
needs further developments? If so, any hints on how to implement that?

Here is the Bro scripts I used (debug prints removed) :

master.bro:

@load base/frameworks/broker
@load broker_wrapper

const broker_port: port = 6666/tcp &redef;
redef exit_only_after_terminate = T;

global h: opaque of Broker::Handle;

event bro_init()
{
    Broker::enable();
    Broker::listen(broker_port, "127.0.0.1");

    h = Broker::create_master("test_store");

    WRAPPER::broker_table_create(h, "test");
}

event Broker::incoming_connection_established(peer_name: string)
{
    print "Conn established: ", peer_name;
}

frontend.bro:

@load base/frameworks/broker
@load broker_wrapper

const broker_port: port = 6666/tcp &redef;
redef exit_only_after_terminate = T;

global h: opaque of Broker::Handle;

event bro_init()
{
    Broker::enable();
    Broker::connect("127.0.0.1", broker_port, 1secs);
}

event Broker::outgoing_connection_established(peer_address: string,
                              peer_port: port,
                        peer_name: string)
{
    print "Conn established: ", peer_address, peer_port, peer_name;

    h = Broker::create_clone("test_store");

    WRAPPER::broker_table_insert(h, "test", Broker::data("one"),
Broker::data("this"));
    WRAPPER::broker_table_insert(h, "test", Broker::data("two"),
Broker::data("is"));
    WRAPPER::broker_table_insert(h, "test", Broker::data("three"),
Broker::data("sparta"));
}

event Broker::outgoing_connection_broken(peer_name: string, peer_port: port)
{
    print "Connection closed by remote peer";
    terminate();
}

broker_wrapper.bro:

module WRAPPER;

export {
       global broker_table_create: function(h: opaque of Broker::Handle,
name: string);
       global broker_table_insert: function(h: opaque of Broker::Handle,
name: string, key: Broker::Data, val: Broker::Data);
}

function broker_table_create(h: opaque of Broker::Handle,
                 name: string)
{
    local tab = Broker::table_create();
    Broker::insert(h, Broker::data(name), tab);
}

function broker_table_insert(h: opaque of Broker::Handle,
                 name: string,
                key: Broker::Data,
                val: Broker::Data)
{
    # look for table with name 'name'
    when (local res = Broker::lookup(h, Broker::data(name)))
    {
        # insert element "key" = "value" into table
        local status = Broker::table_insert(res$result,
                        key,
                        val);

                        print res$result;

        # insert table back into store
        Broker::insert(h, Broker::data(name), res$result);
    }
    timeout 10sec
    {
        print fmt("timeout broker_table_insert: %s key: %s val: %s",
name, key, val);
    }
}

I also did a second test:

If I try to add an element to a set, then test its existence in it with
Broker::set_contains() and it appears not to be in in the set. Any idea why?

Any help is welcome!

- --
Jeffrey BENCTEUX
ANSSI/COSSI/DTO/BSD

2

Hi Bencteux,

You are exactly right: the table operations aren't atomic currently
(i.e., they don't happen server-side). Set manipulations are. The main
difference is that for sets there are explicit store operations for
inserting/removing elements, whereas for tables there are not.

I don't think there's a particular reason that Broker doesn't offer
table operations. We are working on a 2nd-gen Broker API currently
which simplifies usage quite a bit overall. It doesn't have table
operations either yet (nor set operations in fact) but I'm planing on
adding them shortly. If you want to track progress there, it's in the
topic/actor-framework branch. This work shouldn't take very long
anymore to get ready.

Robin

3

I stand corrected: it actually did have them already, the API just
made it easy to miss. I've now added explicit methods for
set/table/vector manipulations to the store API in that branch branch
(Note that they aren't accessible from the Bro side yet, that'll come
shortly).

Robin