Changeset 740 for vendor/current/source4/wrepl_server

Timestamp:

Nov 14, 2012, 12:59:34 PM (13 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to 3.6.0

Location:

vendor/current/source4/wrepl_server

Files:

• config.mk (deleted)
• wrepl_apply_records.c (modified) (17 diffs)
• wrepl_in_call.c (modified) (8 diffs)
• wrepl_in_connection.c (modified) (15 diffs)
• wrepl_out_helpers.c (modified) (27 diffs)
• wrepl_scavenging.c (modified) (8 diffs)
• wrepl_server.c (modified) (8 diffs)
• wrepl_server.h (modified) (3 diffs)
• wscript_build (added)

vendor/current/source4/wrepl_server/wrepl_apply_records.c

@@ -21 +21 @@
 
 #include "includes.h"
+#include <tevent.h>
 #include "smbd/service_task.h"
 #include "lib/messaging/irpc.h"
-#include "librpc/gen_ndr/ndr_irpc.h"
+#include "librpc/gen_ndr/ndr_irpc_c.h"
 #include "librpc/gen_ndr/ndr_winsrepl.h"
 #include "wrepl_server/wrepl_server.h"
@@ -883 +884 @@
 
 struct r_do_challenge_state {
-        struct messaging_context *msg_ctx;
+        struct dcerpc_binding_handle *irpc_handle;
         struct wreplsrv_partner *partner;
         struct winsdb_record *rec;
@@ -889 +890 @@
         struct wrepl_name replica;
         struct nbtd_proxy_wins_challenge r;
+        struct nbtd_proxy_wins_release_demand dr;
 };
 
-static void r_do_late_release_demand_handler(struct irpc_request *ireq)
+static void r_do_late_release_demand_handler(struct tevent_req *subreq)
 {
         NTSTATUS status;
-        struct r_do_challenge_state *state = talloc_get_type(ireq->async.private_data,
-                                                             struct r_do_challenge_state);
-
-        status = irpc_call_recv(ireq);
+        struct r_do_challenge_state *state =
+                tevent_req_callback_data(subreq,
+                struct r_do_challenge_state);
+
+        status = dcerpc_nbtd_proxy_wins_release_demand_r_recv(subreq, state);
+        TALLOC_FREE(subreq);
+
         /* don't care about the result */
         talloc_free(state);
@@ -904 +909 @@
 static NTSTATUS r_do_late_release_demand(struct r_do_challenge_state *state)
 {
-        struct irpc_request *ireq;
-        struct server_id *nbt_servers;
-        struct nbtd_proxy_wins_release_demand r;
+        struct tevent_req *subreq;
         uint32_t i;
 
@@ -912 +915 @@
                  nbt_name_string(state, &state->replica.name)));
 
-        nbt_servers = irpc_servers_byname(state->msg_ctx, state, "nbt_server");
-        if ((nbt_servers == NULL) || (nbt_servers[0].id == 0)) {
-                return NT_STATUS_INTERNAL_ERROR;
-        }
-
-        r.in.name       = state->replica.name;
-        r.in.num_addrs  = state->r.out.num_addrs;
-        r.in.addrs      = talloc_array(state, struct nbtd_proxy_wins_addr, r.in.num_addrs);
-        NT_STATUS_HAVE_NO_MEMORY(r.in.addrs);
+        state->dr.in.name       = state->replica.name;
+        state->dr.in.num_addrs  = state->r.out.num_addrs;
+        state->dr.in.addrs      = talloc_array(state,
+                                               struct nbtd_proxy_wins_addr,
+                                               state->dr.in.num_addrs);
+        NT_STATUS_HAVE_NO_MEMORY(state->dr.in.addrs);
         /* TODO: fix pidl to handle inline ipv4address arrays */
-        for (i=0; i < r.in.num_addrs; i++) {
-                r.in.addrs[i].addr = state->r.out.addrs[i].addr;
-        }
-
-        ireq = IRPC_CALL_SEND(state->msg_ctx, nbt_servers[0],
-                              irpc, NBTD_PROXY_WINS_RELEASE_DEMAND,
-                              &r, state);
-        NT_STATUS_HAVE_NO_MEMORY(ireq);
-
-        ireq->async.fn          = r_do_late_release_demand_handler;
-        ireq->async.private_data= state;
+        for (i=0; i < state->dr.in.num_addrs; i++) {
+                state->dr.in.addrs[i].addr = state->r.out.addrs[i].addr;
+        }
+
+        subreq = dcerpc_nbtd_proxy_wins_release_demand_r_send(state,
+                        state->partner->service->task->event_ctx,
+                        state->irpc_handle,
+                        &state->dr);
+        NT_STATUS_HAVE_NO_MEMORY(subreq);
+
+        tevent_req_set_callback(subreq, r_do_late_release_demand_handler, state);
 
         return NT_STATUS_OK;
@@ -952 +952 @@
 _UA_MA_DI_A<00>: C:BEST vs. B:BEST2 (C:ALL) => B:MHOMED => MHOMED_MERGE
 */
-static void r_do_challenge_handler(struct irpc_request *ireq)
+static void r_do_challenge_handler(struct tevent_req *subreq)
 {
         NTSTATUS status;
-        struct r_do_challenge_state *state = talloc_get_type(ireq->async.private_data,
-                                                             struct r_do_challenge_state);
+        struct r_do_challenge_state *state =
+                tevent_req_callback_data(subreq,
+                struct r_do_challenge_state);
         bool old_is_subset = false;
         bool new_is_subset = false;
@@ -963 +964 @@
         uint32_t num_rec_addrs;
 
-        status = irpc_call_recv(ireq);
+        status = dcerpc_nbtd_proxy_wins_challenge_r_recv(subreq, state);
+        TALLOC_FREE(subreq);
 
         DEBUG(4,("r_do_challenge_handler: %s: %s\n", 
@@ -1034 +1036 @@
                                struct wrepl_name *replica)
 {
-        struct irpc_request *ireq;
         struct r_do_challenge_state *state;
-        struct server_id *nbt_servers;
+        struct tevent_req *subreq;
         const char **addrs;
         uint32_t i;
@@ -1045 +1046 @@
         state = talloc_zero(mem_ctx, struct r_do_challenge_state);
         NT_STATUS_HAVE_NO_MEMORY(state);
-        state->msg_ctx  = partner->service->task->msg_ctx;
         state->partner  = partner;
         state->rec      = talloc_steal(state, rec);
@@ -1055 +1055 @@
         talloc_steal(state, replica->addresses);
 
-        nbt_servers = irpc_servers_byname(state->msg_ctx, state, "nbt_server");
-        if ((nbt_servers == NULL) || (nbt_servers[0].id == 0)) {
+        state->irpc_handle = irpc_binding_handle_by_name(state,
+                                                         partner->service->task->msg_ctx,
+                                                         "nbt_server",
+                                                         &ndr_table_irpc);
+        if (state->irpc_handle == NULL) {
                 return NT_STATUS_INTERNAL_ERROR;
         }
@@ -1071 +1074 @@
         }
 
-        ireq = IRPC_CALL_SEND(state->msg_ctx, nbt_servers[0],
-                              irpc, NBTD_PROXY_WINS_CHALLENGE,
-                              &state->r, state);
-        NT_STATUS_HAVE_NO_MEMORY(ireq);
-
-        ireq->async.fn          = r_do_challenge_handler;
-        ireq->async.private_data= state;
+        subreq = dcerpc_nbtd_proxy_wins_challenge_r_send(state,
+                        state->partner->service->task->event_ctx,
+                        state->irpc_handle,
+                        &state->r);
+        NT_STATUS_HAVE_NO_MEMORY(subreq);
+
+        tevent_req_set_callback(subreq, r_do_challenge_handler, state);
 
         talloc_steal(partner, state);
@@ -1084 +1087 @@
 
 struct r_do_release_demand_state {
-        struct messaging_context *msg_ctx;
         struct nbtd_proxy_wins_release_demand r;
 };
 
-static void r_do_release_demand_handler(struct irpc_request *ireq)
+static void r_do_release_demand_handler(struct tevent_req *subreq)
 {
         NTSTATUS status;
-        struct r_do_release_demand_state *state = talloc_get_type(ireq->async.private_data,
-                                                  struct r_do_release_demand_state);
-
-        status = irpc_call_recv(ireq);
+        struct r_do_release_demand_state *state =
+                tevent_req_callback_data(subreq,
+                struct r_do_release_demand_state);
+
+        status = dcerpc_nbtd_proxy_wins_release_demand_r_recv(subreq, state);
+        TALLOC_FREE(subreq);
+
         /* don't care about the result */
         talloc_free(state);
@@ -1106 +1111 @@
 {
         NTSTATUS status;
-        struct irpc_request *ireq;
-        struct server_id *nbt_servers;
+        struct dcerpc_binding_handle *irpc_handle;
         const char **addrs;
         struct winsdb_addr **addresses;
         struct r_do_release_demand_state *state;
+        struct tevent_req *subreq;
         uint32_t i;
 
@@ -1128 +1133 @@
         state = talloc_zero(mem_ctx, struct r_do_release_demand_state);
         NT_STATUS_HAVE_NO_MEMORY(state);
-        state->msg_ctx  = partner->service->task->msg_ctx;
-
-        nbt_servers = irpc_servers_byname(state->msg_ctx, state, "nbt_server");
-        if ((nbt_servers == NULL) || (nbt_servers[0].id == 0)) {
+
+        irpc_handle = irpc_binding_handle_by_name(state,
+                                                  partner->service->task->msg_ctx,
+                                                  "nbt_server",
+                                                  &ndr_table_irpc);
+        if (irpc_handle == NULL) {
                 return NT_STATUS_INTERNAL_ERROR;
         }
@@ -1147 +1154 @@
         }
 
-        ireq = IRPC_CALL_SEND(state->msg_ctx, nbt_servers[0],
-                              irpc, NBTD_PROXY_WINS_RELEASE_DEMAND,
-                              &state->r, state);
-        NT_STATUS_HAVE_NO_MEMORY(ireq);
-
-        ireq->async.fn          = r_do_release_demand_handler;
-        ireq->async.private_data= state;
+        subreq = dcerpc_nbtd_proxy_wins_release_demand_r_send(state,
+                        partner->service->task->event_ctx,
+                        irpc_handle,
+                        &state->r);
+        NT_STATUS_HAVE_NO_MEMORY(subreq);
+
+        tevent_req_set_callback(subreq, r_do_release_demand_handler, state);
 
         talloc_steal(partner, state);
@@ -1194 +1201 @@
         bool skip_replica_owned_by_us = false;
         bool become_owner = true;
-        bool propagate = lp_parm_bool(partner->service->task->lp_ctx, NULL, "wreplsrv", "propagate name releases", false);
+        bool propagate = lpcfg_parm_bool(partner->service->task->lp_ctx, NULL, "wreplsrv", "propagate name releases", false);
         const char *local_owner = partner->service->wins_db->local_owner;
 
@@ -1358 +1365 @@
         bool replica_vs_replica = false;
         bool local_vs_replica = false;
+
+        if (replica->name.scope) {
+                TALLOC_CTX *parent;
+                const char *scope;
+
+                /*
+                 * Windows 2008 truncates the scope to 237 bytes,
+                 * so we do...
+                 */
+                parent = talloc_parent(replica->name.scope);
+                scope = talloc_strndup(parent, replica->name.scope, 237);
+                NT_STATUS_HAVE_NO_MEMORY(scope);
+                replica->name.scope = scope;
+        }
 
         status = winsdb_lookup(partner->service->wins_db,

vendor/current/source4/wrepl_server/wrepl_in_call.c

@@ -22 +22 @@
 #include "includes.h"
 #include "lib/events/events.h"
-#include "lib/socket/socket.h"
+#include "lib/tsocket/tsocket.h"
+#include "smbd/service_task.h"
 #include "smbd/service_stream.h"
 #include "libcli/wrepl/winsrepl.h"
@@ -28 +29 @@
 #include "libcli/composite/composite.h"
 #include "nbt_server/wins/winsdb.h"
-#include "lib/ldb/include/ldb.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb.h>
+#include <ldb_errors.h>
 #include "system/time.h"
+#include "lib/util/tsort.h"
+#include "param/param.h"
 
 static NTSTATUS wreplsrv_in_start_association(struct wreplsrv_in_call *call)
@@ -199 +202 @@
         struct winsdb_record *rec;
         NTSTATUS status;
-        uint32_t i, j;
+        unsigned int i, j;
         time_t now = time(NULL);
 
@@ -301 +304 @@
 
         /* sort the names before we send them */
-        qsort(names, j, sizeof(struct wrepl_wins_name), (comparison_fn_t)wreplsrv_in_sort_wins_name);
+        TYPESAFE_QSORT(names, j, wreplsrv_in_sort_wins_name);
 
         DEBUG(2,("WINSREPL:reply [%u] records owner[%s] min[%llu] max[%llu] to partner[%s]\n",
@@ -341 +344 @@
         struct wrepl_table *update_in = &call->req_packet.message.replication.info.table;
         struct wreplsrv_in_update_state *update_state;
-        uint16_t fde_flags;
+        NTSTATUS status;
 
         DEBUG(2,("WREPL_REPL_UPDATE: partner[%s] initiator[%s] num_owners[%u]\n",
                 call->wreplconn->partner->address,
                 update_in->initiator, update_in->partner_count));
-
-        /* 
-         * we need to flip the connection into a client connection
-         * and do a WREPL_REPL_SEND_REQUEST's on the that connection
-         * and then stop this connection
-         */
-        fde_flags = event_get_fd_flags(wrepl_in->conn->event.fde);
-        talloc_free(wrepl_in->conn->event.fde);
-        wrepl_in->conn->event.fde = NULL;
 
         update_state = talloc(wrepl_in, struct wreplsrv_in_update_state);
@@ -365 +359 @@
         wrepl_out->assoc_ctx.our_ctx    = wrepl_in->assoc_ctx.our_ctx;
         wrepl_out->assoc_ctx.peer_ctx   = wrepl_in->assoc_ctx.peer_ctx;
-        wrepl_out->sock                 = wrepl_socket_merge(wrepl_out,
-                                                             wrepl_in->conn->event.ctx,
-                                                             wrepl_in->conn->socket,
-                                                             wrepl_in->packet);
-        NT_STATUS_HAVE_NO_MEMORY(wrepl_out->sock);
-
-        event_set_fd_flags(wrepl_out->sock->event.fde, fde_flags);
+        wrepl_out->sock                 = wrepl_socket_init(wrepl_out,
+                                                            wrepl_in->conn->event.ctx);
+                                                            
+        NT_STATUS_HAVE_NO_MEMORY_AND_FREE(wrepl_out->sock, update_state);
+
+        TALLOC_FREE(wrepl_in->send_queue);
+
+        status = wrepl_socket_donate_stream(wrepl_out->sock, &wrepl_in->tstream);
+        NT_STATUS_NOT_OK_RETURN_AND_FREE(status, update_state);
 
         update_state->wrepl_in                  = wrepl_in;
@@ -382 +378 @@
         update_state->creq = wreplsrv_pull_cycle_send(update_state, &update_state->cycle_io);
         if (!update_state->creq) {
+                talloc_free(update_state);
                 return NT_STATUS_INTERNAL_ERROR;
         }
@@ -433 +430 @@
 
         if (!call->wreplconn->partner) {
-                struct socket_address *partner_ip = socket_get_peer_addr(call->wreplconn->conn->socket, call);
-
-                call->wreplconn->partner = wreplsrv_find_partner(call->wreplconn->service, partner_ip->addr);
+                struct tsocket_address *peer_addr = call->wreplconn->conn->remote_address;
+                char *peer_ip;
+
+                if (!tsocket_address_is_inet(peer_addr, "ipv4")) {
+                        DEBUG(0,("wreplsrv_in_replication: non ipv4 peer addr '%s'\n",
+                                tsocket_address_string(peer_addr, call)));
+                        return NT_STATUS_INTERNAL_ERROR;
+                }
+
+                peer_ip = tsocket_address_inet_addr_string(peer_addr, call);
+                if (peer_ip == NULL) {
+                        return NT_STATUS_NO_MEMORY;
+                }
+
+                call->wreplconn->partner = wreplsrv_find_partner(call->wreplconn->service, peer_ip);
                 if (!call->wreplconn->partner) {
-                        DEBUG(1,("Failing WINS replication from non-partner %s\n",
-                                 partner_ip ? partner_ip->addr : NULL));
+                        DEBUG(1,("Failing WINS replication from non-partner %s\n", peer_ip));
                         return wreplsrv_in_stop_assoc_ctx(call);
                 }

vendor/current/source4/wrepl_server/wrepl_in_connection.c

@@ -32 +32 @@
 #include "system/network.h"
 #include "lib/socket/netif.h"
+#include "lib/tsocket/tsocket.h"
+#include "libcli/util/tstream.h"
 #include "param/param.h"
 
@@ -39 +41 @@
 }
 
-static int terminate_after_send_destructor(struct wreplsrv_in_connection **tas)
-{
-        wreplsrv_terminate_in_connection(*tas, "wreplsrv_in_connection: terminate_after_send");
-        return 0;
-}
-
 /*
   receive some data on a WREPL connection
 */
-static NTSTATUS wreplsrv_recv_request(void *private_data, DATA_BLOB blob)
-{
-        struct wreplsrv_in_connection *wreplconn = talloc_get_type(private_data, struct wreplsrv_in_connection);
-        struct wreplsrv_in_call *call;
-        DATA_BLOB packet_in_blob;
-        DATA_BLOB packet_out_blob;
+static NTSTATUS wreplsrv_process(struct wreplsrv_in_connection *wrepl_conn,
+                                 struct wreplsrv_in_call **_call)
+{
         struct wrepl_wrap packet_out_wrap;
         NTSTATUS status;
         enum ndr_err_code ndr_err;
-
-        call = talloc_zero(wreplconn, struct wreplsrv_in_call);
-        NT_STATUS_HAVE_NO_MEMORY(call);
-        call->wreplconn = wreplconn;
-        talloc_steal(call, blob.data);
-
-        packet_in_blob.data = blob.data + 4;
-        packet_in_blob.length = blob.length - 4;
-
-        ndr_err = ndr_pull_struct_blob(&packet_in_blob, call, 
-                                       lp_iconv_convenience(wreplconn->service->task->lp_ctx),
+        struct wreplsrv_in_call *call = *_call;
+
+        ndr_err = ndr_pull_struct_blob(&call->in, call,
                                        &call->req_packet,
                                        (ndr_pull_flags_fn_t)ndr_pull_wrepl_packet);
@@ -75 +60 @@
 
         if (DEBUGLVL(10)) {
-                DEBUG(10,("Received WINS-Replication packet of length %u\n", 
-                          (unsigned)packet_in_blob.length + 4));
+                DEBUG(10,("Received WINS-Replication packet of length %u\n",
+                          (unsigned int) call->in.length + 4));
                 NDR_PRINT_DEBUG(wrepl_packet, &call->req_packet);
         }
@@ -85 +70 @@
                 /* w2k just ignores invalid packets, so we do */
                 DEBUG(10,("Received WINS-Replication packet was invalid, we just ignore it\n"));
-                talloc_free(call);
+                TALLOC_FREE(call);
+                *_call = NULL;
                 return NT_STATUS_OK;
         }
@@ -91 +77 @@
         /* and now encode the reply */
         packet_out_wrap.packet = call->rep_packet;
-        ndr_err = ndr_push_struct_blob(&packet_out_blob, call, 
-                                       lp_iconv_convenience(wreplconn->service->task->lp_ctx),
+        ndr_err = ndr_push_struct_blob(&call->out, call,
                                        &packet_out_wrap,
-                                      (ndr_push_flags_fn_t)ndr_push_wrepl_wrap);
+                                       (ndr_push_flags_fn_t) ndr_push_wrepl_wrap);
         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                 return ndr_map_error2ntstatus(ndr_err);
@@ -100 +85 @@
 
         if (DEBUGLVL(10)) {
-                DEBUG(10,("Sending WINS-Replication packet of length %d\n", (int)packet_out_blob.length));
+                DEBUG(10,("Sending WINS-Replication packet of length %u\n",
+                         (unsigned int) call->out.length));
                 NDR_PRINT_DEBUG(wrepl_packet, &call->rep_packet);
         }
 
+        return NT_STATUS_OK;
+}
+
+static void wreplsrv_call_loop(struct tevent_req *subreq);
+
+/*
+  called when we get a new connection
+*/
+static void wreplsrv_accept(struct stream_connection *conn)
+{
+        struct wreplsrv_service *service = talloc_get_type(conn->private_data, struct wreplsrv_service);
+        struct wreplsrv_in_connection *wrepl_conn;
+        struct tsocket_address *peer_addr;
+        char *peer_ip;
+        struct tevent_req *subreq;
+        int rc;
+
+        wrepl_conn = talloc_zero(conn, struct wreplsrv_in_connection);
+        if (wrepl_conn == NULL) {
+                stream_terminate_connection(conn,
+                                            "wreplsrv_accept: out of memory");
+                return;
+        }
+
+        wrepl_conn->send_queue = tevent_queue_create(conn, "wrepl_accept");
+        if (wrepl_conn->send_queue == NULL) {
+                stream_terminate_connection(conn,
+                                            "wrepl_accept: out of memory");
+                return;
+        }
+
+        TALLOC_FREE(conn->event.fde);
+
+        rc = tstream_bsd_existing_socket(wrepl_conn,
+                                         socket_get_fd(conn->socket),
+                                         &wrepl_conn->tstream);
+        if (rc < 0) {
+                stream_terminate_connection(conn,
+                                            "wrepl_accept: out of memory");
+                return;
+        }
+        socket_set_flags(conn->socket, SOCKET_FLAG_NOCLOSE);
+
+        wrepl_conn->conn = conn;
+        wrepl_conn->service = service;
+
+        peer_addr = conn->remote_address;
+
+        if (!tsocket_address_is_inet(peer_addr, "ipv4")) {
+                DEBUG(0,("wreplsrv_accept: non ipv4 peer addr '%s'\n",
+                        tsocket_address_string(peer_addr, wrepl_conn)));
+                wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_accept: "
+                                "invalid peer IP");
+                return;
+        }
+
+        peer_ip = tsocket_address_inet_addr_string(peer_addr, wrepl_conn);
+        if (peer_ip == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_accept: "
+                                "could not convert peer IP into a string");
+                return;
+        }
+
+        wrepl_conn->partner = wreplsrv_find_partner(service, peer_ip);
+        irpc_add_name(conn->msg_ctx, "wreplsrv_connection");
+
+        /*
+         * The wrepl pdu's has the length as 4 byte (initial_read_size),
+         * packet_full_request_u32 provides the pdu length then.
+         */
+        subreq = tstream_read_pdu_blob_send(wrepl_conn,
+                                            wrepl_conn->conn->event.ctx,
+                                            wrepl_conn->tstream,
+                                            4, /* initial_read_size */
+                                            packet_full_request_u32,
+                                            wrepl_conn);
+        if (subreq == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_accept: "
+                                "no memory for tstream_read_pdu_blob_send");
+                return;
+        }
+        tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_conn);
+}
+
+static void wreplsrv_call_writev_done(struct tevent_req *subreq);
+
+static void wreplsrv_call_loop(struct tevent_req *subreq)
+{
+        struct wreplsrv_in_connection *wrepl_conn = tevent_req_callback_data(subreq,
+                                      struct wreplsrv_in_connection);
+        struct wreplsrv_in_call *call;
+        NTSTATUS status;
+
+        call = talloc_zero(wrepl_conn, struct wreplsrv_in_call);
+        if (call == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
+                                "no memory for wrepl_samba3_call");
+                return;
+        }
+        call->wreplconn = wrepl_conn;
+
+        status = tstream_read_pdu_blob_recv(subreq,
+                                            call,
+                                            &call->in);
+        TALLOC_FREE(subreq);
+        if (!NT_STATUS_IS_OK(status)) {
+                const char *reason;
+
+                reason = talloc_asprintf(call, "wreplsrv_call_loop: "
+                                         "tstream_read_pdu_blob_recv() - %s",
+                                         nt_errstr(status));
+                if (!reason) {
+                        reason = nt_errstr(status);
+                }
+
+                wreplsrv_terminate_in_connection(wrepl_conn, reason);
+                return;
+        }
+
+        DEBUG(10,("Received wrepl packet of length %lu from %s\n",
+                 (long) call->in.length,
+                 tsocket_address_string(wrepl_conn->conn->remote_address, call)));
+
+        /* skip length header */
+        call->in.data += 4;
+        call->in.length -= 4;
+
+        status = wreplsrv_process(wrepl_conn, &call);
+        if (!NT_STATUS_IS_OK(status)) {
+                const char *reason;
+
+                reason = talloc_asprintf(call, "wreplsrv_call_loop: "
+                                         "tstream_read_pdu_blob_recv() - %s",
+                                         nt_errstr(status));
+                if (reason == NULL) {
+                        reason = nt_errstr(status);
+                }
+
+                wreplsrv_terminate_in_connection(wrepl_conn, reason);
+                return;
+        }
+
+        /* We handed over the connection so we're done here */
+        if (wrepl_conn->tstream == NULL) {
+            return;
+        }
+
+        /* Invalid WINS-Replication packet, we just ignore it */
+        if (call == NULL) {
+                goto noreply;
+        }
+
+        call->out_iov[0].iov_base = (char *) call->out.data;
+        call->out_iov[0].iov_len = call->out.length;
+
+        subreq = tstream_writev_queue_send(call,
+                                           wrepl_conn->conn->event.ctx,
+                                           wrepl_conn->tstream,
+                                           wrepl_conn->send_queue,
+                                           call->out_iov, 1);
+        if (subreq == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
+                                "no memory for tstream_writev_queue_send");
+                return;
+        }
+        tevent_req_set_callback(subreq, wreplsrv_call_writev_done, call);
+
+noreply:
+        /*
+         * The wrepl pdu's has the length as 4 byte (initial_read_size),
+         *  provides the pdu length then.
+         */
+        subreq = tstream_read_pdu_blob_send(wrepl_conn,
+                                            wrepl_conn->conn->event.ctx,
+                                            wrepl_conn->tstream,
+                                            4, /* initial_read_size */
+                                            packet_full_request_u32,
+                                            wrepl_conn);
+        if (subreq == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
+                                "no memory for tstream_read_pdu_blob_send");
+                return;
+        }
+        tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_conn);
+}
+
+static void wreplsrv_call_writev_done(struct tevent_req *subreq)
+{
+        struct wreplsrv_in_call *call = tevent_req_callback_data(subreq,
+                        struct wreplsrv_in_call);
+        int sys_errno;
+        int rc;
+
+        rc = tstream_writev_queue_recv(subreq, &sys_errno);
+        TALLOC_FREE(subreq);
+        if (rc == -1) {
+                const char *reason;
+
+                reason = talloc_asprintf(call, "wreplsrv_call_writev_done: "
+                                         "tstream_writev_queue_recv() - %d:%s",
+                                         sys_errno, strerror(sys_errno));
+                if (reason == NULL) {
+                        reason = "wreplsrv_call_writev_done: "
+                                 "tstream_writev_queue_recv() failed";
+                }
+
+                wreplsrv_terminate_in_connection(call->wreplconn, reason);
+                return;
+        }
+
         if (call->terminate_after_send) {
-                struct wreplsrv_in_connection **tas;
-                tas = talloc(packet_out_blob.data, struct wreplsrv_in_connection *);
-                NT_STATUS_HAVE_NO_MEMORY(tas);
-                *tas = wreplconn;
-                talloc_set_destructor(tas, terminate_after_send_destructor);
-        }
-
-        status = packet_send(wreplconn->packet, packet_out_blob);
-        NT_STATUS_NOT_OK_RETURN(status);
+                wreplsrv_terminate_in_connection(call->wreplconn,
+                                "wreplsrv_in_connection: terminate_after_send");
+                return;
+        }
 
         talloc_free(call);
-        return NT_STATUS_OK;
-}
-
-/*
-  called when the socket becomes readable
+}
+
+/*
+  called on a tcp recv
 */
 static void wreplsrv_recv(struct stream_connection *conn, uint16_t flags)
 {
-        struct wreplsrv_in_connection *wreplconn = talloc_get_type(conn->private_data,
-                                                                   struct wreplsrv_in_connection);
-
-        packet_recv(wreplconn->packet);
-}
-
-/*
-  called when the socket becomes writable
+        struct wreplsrv_in_connection *wrepl_conn = talloc_get_type(conn->private_data,
+                                                        struct wreplsrv_in_connection);
+        /* this should never be triggered! */
+        DEBUG(0,("Terminating connection - '%s'\n", "wrepl_recv: called"));
+        wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_recv: called");
+}
+
+/*
+  called when we can write to a connection
 */
 static void wreplsrv_send(struct stream_connection *conn, uint16_t flags)
 {
-        struct wreplsrv_in_connection *wreplconn = talloc_get_type(conn->private_data,
-                                                                   struct wreplsrv_in_connection);
-        packet_queue_run(wreplconn->packet);
-}
-
-/*
-  handle socket recv errors
-*/
-static void wreplsrv_recv_error(void *private_data, NTSTATUS status)
-{
-        struct wreplsrv_in_connection *wreplconn = talloc_get_type(private_data,
-                                                                   struct wreplsrv_in_connection);
-        wreplsrv_terminate_in_connection(wreplconn, nt_errstr(status));
-}
-
-/*
-  called when we get a new connection
-*/
-static void wreplsrv_accept(struct stream_connection *conn)
-{
-        struct wreplsrv_service *service = talloc_get_type(conn->private_data, struct wreplsrv_service);
-        struct wreplsrv_in_connection *wreplconn;
-        struct socket_address *peer_ip;
-
-        wreplconn = talloc_zero(conn, struct wreplsrv_in_connection);
-        if (!wreplconn) {
-                stream_terminate_connection(conn, "wreplsrv_accept: out of memory");
-                return;
-        }
-
-        wreplconn->packet = packet_init(wreplconn);
-        if (!wreplconn->packet) {
-                wreplsrv_terminate_in_connection(wreplconn, "wreplsrv_accept: out of memory");
-                return;
-        }
-        packet_set_private(wreplconn->packet, wreplconn);
-        packet_set_socket(wreplconn->packet, conn->socket);
-        packet_set_callback(wreplconn->packet, wreplsrv_recv_request);
-        packet_set_full_request(wreplconn->packet, packet_full_request_u32);
-        packet_set_error_handler(wreplconn->packet, wreplsrv_recv_error);
-        packet_set_event_context(wreplconn->packet, conn->event.ctx);
-        packet_set_fde(wreplconn->packet, conn->event.fde);
-        packet_set_serialise(wreplconn->packet);
-
-        wreplconn->conn         = conn;
-        wreplconn->service      = service;
-
-        peer_ip = socket_get_peer_addr(conn->socket, wreplconn);
-        if (!peer_ip) {
-                wreplsrv_terminate_in_connection(wreplconn, "wreplsrv_accept: could not obtain peer IP from kernel");
-                return;
-        }
-
-        wreplconn->partner      = wreplsrv_find_partner(service, peer_ip->addr);
-
-        conn->private_data = wreplconn;
-
-        irpc_add_name(conn->msg_ctx, "wreplsrv_connection");
+        struct wreplsrv_in_connection *wrepl_conn = talloc_get_type(conn->private_data,
+                                                        struct wreplsrv_in_connection);
+        /* this should never be triggered! */
+        DEBUG(0,("Terminating connection - '%s'\n", "wrepl_send: called"));
+        wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_send: called");
 }
 
@@ -206 +344 @@
 */
 NTSTATUS wreplsrv_in_connection_merge(struct wreplsrv_partner *partner,
-                                      struct socket_context *sock,
-                                      struct packet_context *packet,
+                                      uint32_t peer_assoc_ctx,
+                                      struct tstream_context **stream,
                                       struct wreplsrv_in_connection **_wrepl_in)
 {
@@ -214 +352 @@
         const struct model_ops *model_ops;
         struct stream_connection *conn;
+        struct tevent_req *subreq;
         NTSTATUS status;
 
@@ -219 +358 @@
            ask for the single process model ops and pass these to the
            stream_setup_socket() call. */
-        model_ops = process_model_startup(service->task->event_ctx, "single");
+        model_ops = process_model_startup("single");
         if (!model_ops) {
                 DEBUG(0,("Can't find 'single' process model_ops"));
@@ -230 +369 @@
         wrepl_in->service       = service;
         wrepl_in->partner       = partner;
-
-        status = stream_new_connection_merge(service->task->event_ctx, service->task->lp_ctx, model_ops,
-                                             sock, &wreplsrv_stream_ops, service->task->msg_ctx,
-                                             wrepl_in, &conn);
+        wrepl_in->tstream       = talloc_move(wrepl_in, stream);
+        wrepl_in->assoc_ctx.peer_ctx = peer_assoc_ctx;
+
+        status = stream_new_connection_merge(service->task->event_ctx,
+                                             service->task->lp_ctx,
+                                             model_ops,
+                                             &wreplsrv_stream_ops,
+                                             service->task->msg_ctx,
+                                             wrepl_in,
+                                             &conn);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -243 +388 @@
         talloc_steal(conn, wrepl_in);
 
+        wrepl_in->send_queue = tevent_queue_create(wrepl_in, "wreplsrv_in_connection_merge");
+        if (wrepl_in->send_queue == NULL) {
+                stream_terminate_connection(conn,
+                                            "wreplsrv_in_connection_merge: out of memory");
+                return NT_STATUS_NO_MEMORY;
+        }
+
         /*
-         * now update the packet handling callback,...
+         * The wrepl pdu's has the length as 4 byte (initial_read_size),
+         * packet_full_request_u32 provides the pdu length then.
          */
-        wrepl_in->packet        = talloc_steal(wrepl_in, packet);
-        packet_set_private(wrepl_in->packet, wrepl_in);
-        packet_set_socket(wrepl_in->packet, conn->socket);
-        packet_set_callback(wrepl_in->packet, wreplsrv_recv_request);
-        packet_set_full_request(wrepl_in->packet, packet_full_request_u32);
-        packet_set_error_handler(wrepl_in->packet, wreplsrv_recv_error);
-        packet_set_event_context(wrepl_in->packet, conn->event.ctx);
-        packet_set_fde(wrepl_in->packet, conn->event.fde);
-        packet_set_serialise(wrepl_in->packet);
+        subreq = tstream_read_pdu_blob_send(wrepl_in,
+                                            wrepl_in->conn->event.ctx,
+                                            wrepl_in->tstream,
+                                            4, /* initial_read_size */
+                                            packet_full_request_u32,
+                                            wrepl_in);
+        if (subreq == NULL) {
+                wreplsrv_terminate_in_connection(wrepl_in, "wreplsrv_in_connection_merge: "
+                                "no memory for tstream_read_pdu_blob_send");
+                return NT_STATUS_NO_MEMORY;
+        }
+        tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_in);
 
         *_wrepl_in = wrepl_in;
+
         return NT_STATUS_OK;
 }
@@ -274 +431 @@
            ask for the single process model ops and pass these to the
            stream_setup_socket() call. */
-        model_ops = process_model_startup(task->event_ctx, "single");
+        model_ops = process_model_startup("single");
         if (!model_ops) {
                 DEBUG(0,("Can't find 'single' process model_ops"));
@@ -280 +437 @@
         }
 
-        if (lp_interfaces(lp_ctx) && lp_bind_interfaces_only(lp_ctx)) {
+        if (lpcfg_interfaces(lp_ctx) && lpcfg_bind_interfaces_only(lp_ctx)) {
                 int num_interfaces;
                 int i;
                 struct interface *ifaces;
 
-                load_interfaces(task, lp_interfaces(lp_ctx), &ifaces);
+                load_interfaces(task, lpcfg_interfaces(lp_ctx), &ifaces);
 
                 num_interfaces = iface_count(ifaces);
@@ -295 +452 @@
                 for(i = 0; i < num_interfaces; i++) {
                         address = iface_n_ip(ifaces, i);
-                        status = stream_setup_socket(task->event_ctx, 
-                                                     task->lp_ctx, model_ops, 
+                        status = stream_setup_socket(task, task->event_ctx,
+                                                     task->lp_ctx, model_ops,
                                                      &wreplsrv_stream_ops,
                                                      "ipv4", address, &port, 
-                                                      lp_socket_options(task->lp_ctx), 
+                                                      lpcfg_socket_options(task->lp_ctx),
                                                      service);
                         if (!NT_STATUS_IS_OK(status)) {
@@ -308 +465 @@
                 }
         } else {
-                address = lp_socket_address(lp_ctx);
-                status = stream_setup_socket(task->event_ctx, task->lp_ctx, 
+                address = lpcfg_socket_address(lp_ctx);
+                status = stream_setup_socket(task, task->event_ctx, task->lp_ctx,
                                              model_ops, &wreplsrv_stream_ops,
-                                             "ipv4", address, &port, lp_socket_options(task->lp_ctx), 
+                                             "ipv4", address, &port, lpcfg_socket_options(task->lp_ctx),
                                              service);
                 if (!NT_STATUS_IS_OK(status)) {

vendor/current/source4/wrepl_server/wrepl_out_helpers.c

@@ -42 +42 @@
         enum wreplsrv_out_connect_stage stage;
         struct composite_context *c;
-        struct wrepl_request *req;
-        struct composite_context *c_req;
         struct wrepl_associate assoc_io;
         enum winsrepl_partner_type type;
         struct wreplsrv_out_connection *wreplconn;
+        struct tevent_req *subreq;
 };
 
-static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
-static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
+static void wreplsrv_out_connect_handler_treq(struct tevent_req *subreq);
 
 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
@@ -56 +54 @@
         NTSTATUS status;
 
-        status = wrepl_connect_recv(state->c_req);
-        NT_STATUS_NOT_OK_RETURN(status);
-
-        state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
-        NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-        state->req->async.fn            = wreplsrv_out_connect_handler_req;
-        state->req->async.private_data  = state;
+        status = wrepl_connect_recv(state->subreq);
+        TALLOC_FREE(state->subreq);
+        NT_STATUS_NOT_OK_RETURN(status);
+
+        state->subreq = wrepl_associate_send(state,
+                                             state->wreplconn->service->task->event_ctx,
+                                             state->wreplconn->sock, &state->assoc_io);
+        NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_out_connect_handler_treq,
+                                state);
 
         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
@@ -74 +76 @@
         NTSTATUS status;
 
-        status = wrepl_associate_recv(state->req, &state->assoc_io);
+        status = wrepl_associate_recv(state->subreq, &state->assoc_io);
+        TALLOC_FREE(state->subreq);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -122 +125 @@
 }
 
-static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
-{
-        struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
-                                                   struct wreplsrv_out_connect_state);
-        wreplsrv_out_connect_handler(state);
-        return;
-}
-
-static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
-{
-        struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private_data,
+static void wreplsrv_out_connect_handler_treq(struct tevent_req *subreq)
+{
+        struct wreplsrv_out_connect_state *state = tevent_req_callback_data(subreq,
                                                    struct wreplsrv_out_connect_state);
         wreplsrv_out_connect_handler(state);
@@ -172 +167 @@
         /* we have a connection already, so use it */
         if (wreplconn) {
-                if (!wreplconn->sock->dead) {
+                if (wrepl_socket_is_connected(wreplconn->sock)) {
                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
                         state->wreplconn= wreplconn;
@@ -193 +188 @@
         wreplconn->service      = service;
         wreplconn->partner      = partner;
-        wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
+        wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx);
         if (!wreplconn->sock) goto failed;
 
         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
         state->wreplconn= wreplconn;
-        state->c_req    = wrepl_connect_send(wreplconn->sock,
+        state->subreq   = wrepl_connect_send(state,
+                                             service->task->event_ctx,
+                                             wreplconn->sock,
                                              partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
                                              partner->address);
-        if (!state->c_req) goto failed;
-
-        state->c_req->async.fn                  = wreplsrv_out_connect_handler_creq;
-        state->c_req->async.private_data        = state;
+        if (!state->subreq) goto failed;
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_out_connect_handler_treq,
+                                state);
 
         return c;
@@ -256 +254 @@
         enum wreplsrv_pull_table_stage stage;
         struct composite_context *c;
-        struct wrepl_request *req;
         struct wrepl_pull_table table_io;
         struct wreplsrv_pull_table_io *io;
         struct composite_context *creq;
         struct wreplsrv_out_connection *wreplconn;
+        struct tevent_req *subreq;
 };
 
-static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
+static void wreplsrv_pull_table_handler_treq(struct tevent_req *subreq);
 
 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
@@ -273 +271 @@
 
         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
-        state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
-        NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-        state->req->async.fn            = wreplsrv_pull_table_handler_req;
-        state->req->async.private_data  = state;
+        state->subreq = wrepl_pull_table_send(state,
+                                              state->wreplconn->service->task->event_ctx,
+                                              state->wreplconn->sock, &state->table_io);
+        NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_pull_table_handler_treq,
+                                state);
 
         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
@@ -288 +289 @@
         NTSTATUS status;
 
-        status = wrepl_pull_table_recv(state->req, state, &state->table_io);
+        status = wrepl_pull_table_recv(state->subreq, state, &state->table_io);
+        TALLOC_FREE(state->subreq);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -329 +331 @@
 }
 
-static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
-{
-        struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private_data,
+static void wreplsrv_pull_table_handler_treq(struct tevent_req *subreq)
+{
+        struct wreplsrv_pull_table_state *state = tevent_req_callback_data(subreq,
                                                   struct wreplsrv_pull_table_state);
         wreplsrv_pull_table_handler(state);
@@ -415 +417 @@
         enum wreplsrv_pull_names_stage stage;
         struct composite_context *c;
-        struct wrepl_request *req;
         struct wrepl_pull_names pull_io;
         struct wreplsrv_pull_names_io *io;
         struct composite_context *creq;
         struct wreplsrv_out_connection *wreplconn;
+        struct tevent_req *subreq;
 };
 
-static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
+static void wreplsrv_pull_names_handler_treq(struct tevent_req *subreq);
 
 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
@@ -433 +435 @@
         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
         state->pull_io.in.partner       = state->io->in.owner;
-        state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
-        NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-        state->req->async.fn            = wreplsrv_pull_names_handler_req;
-        state->req->async.private_data  = state;
+        state->subreq = wrepl_pull_names_send(state,
+                                              state->wreplconn->service->task->event_ctx,
+                                              state->wreplconn->sock,
+                                              &state->pull_io);
+        NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_pull_names_handler_treq,
+                                state);
 
         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
@@ -448 +454 @@
         NTSTATUS status;
 
-        status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
+        status = wrepl_pull_names_recv(state->subreq, state, &state->pull_io);
+        TALLOC_FREE(state->subreq);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -489 +496 @@
 }
 
-static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
-{
-        struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private_data,
+static void wreplsrv_pull_names_handler_treq(struct tevent_req *subreq)
+{
+        struct wreplsrv_pull_names_state *state = tevent_req_callback_data(subreq,
                                                   struct wreplsrv_pull_names_state);
         wreplsrv_pull_names_handler(state);
@@ -566 +573 @@
         struct composite_context *creq;
         struct wrepl_associate_stop assoc_stop_io;
-        struct wrepl_request *req;
+        struct tevent_req *subreq;
 };
 
 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
-static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
+static void wreplsrv_pull_cycle_handler_treq(struct tevent_req *subreq);
 
 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
@@ -648 +655 @@
                 state->assoc_stop_io.in.assoc_ctx       = state->io->in.wreplconn->assoc_ctx.peer_ctx;
                 state->assoc_stop_io.in.reason          = 0;
-                state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
-                NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-                state->req->async.fn            = wreplsrv_pull_cycle_handler_req;
-                state->req->async.private_data  = state;
+                state->subreq = wrepl_associate_stop_send(state,
+                                                          state->io->in.wreplconn->service->task->event_ctx,
+                                                          state->io->in.wreplconn->sock,
+                                                          &state->assoc_stop_io);
+                NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+                tevent_req_set_callback(state->subreq,
+                                        wreplsrv_pull_cycle_handler_treq,
+                                        state);
 
                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
@@ -725 +736 @@
         NTSTATUS status;
 
-        status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
+        status = wrepl_associate_stop_recv(state->subreq, &state->assoc_stop_io);
+        TALLOC_FREE(state->subreq);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -772 +784 @@
 }
 
-static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
-{
-        struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private_data,
+static void wreplsrv_pull_cycle_handler_treq(struct tevent_req *subreq)
+{
+        struct wreplsrv_pull_cycle_state *state = tevent_req_callback_data(subreq,
                                                   struct wreplsrv_pull_cycle_state);
         wreplsrv_pull_cycle_handler(state);
@@ -826 +838 @@
 enum wreplsrv_push_notify_stage {
         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
+        WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE,
         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
         WREPLSRV_PUSH_NOTIFY_STAGE_DONE
@@ -837 +850 @@
         bool full_table;
         struct wrepl_send_ctrl ctrl;
-        struct wrepl_request *req;
         struct wrepl_packet req_packet;
         struct wrepl_packet *rep_packet;
         struct composite_context *creq;
         struct wreplsrv_out_connection *wreplconn;
+        struct tevent_req *subreq;
 };
 
 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
-static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
+static void wreplsrv_push_notify_handler_treq(struct tevent_req *subreq);
 
 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
@@ -853 +866 @@
         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
-        struct wreplsrv_in_connection *wrepl_in;
-        NTSTATUS status;
-        struct socket_context *sock;
-        struct packet_context *packet;
-        uint16_t fde_flags;
+        NTSTATUS status;
 
         /* prepare the outgoing request */
@@ -871 +880 @@
 
         /* queue the request */
-        state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
-        NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-        /*
-         * now we need to convert the wrepl_socket (client connection)
-         * into a wreplsrv_in_connection (server connection), because
-         * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
-         * message is received by the peer.
-         */
-
-        /* steal the socket_context */
-        sock = state->wreplconn->sock->sock;
-        state->wreplconn->sock->sock = NULL;
-        talloc_steal(state, sock);
-
-        /* 
-         * steal the packet_context
-         * note the request DATA_BLOB we just send on the
-         * wrepl_socket (client connection) is still unter the 
-         * packet context and will be send to the wire
-         */
-        packet = state->wreplconn->sock->packet;
-        state->wreplconn->sock->packet = NULL;
-        talloc_steal(state, packet);
-
-        /*
-         * get the fde_flags of the old fde event,
-         * so that we can later set the same flags to the new one
-         */
-        fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
-
-        /*
-         * free the wrepl_socket (client connection)
-         */
-        talloc_free(state->wreplconn->sock);
-        state->wreplconn->sock = NULL;
-
-        /*
-         * now create a wreplsrv_in_connection,
-         * on which we act as server
-         *
-         * NOTE: sock and packet will be stolen by
-         *       wreplsrv_in_connection_merge()
-         */
-        status = wreplsrv_in_connection_merge(state->io->in.partner,
-                                              sock, packet, &wrepl_in);
-        NT_STATUS_NOT_OK_RETURN(status);
-
-        event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
-
-        wrepl_in->assoc_ctx.peer_ctx    = state->wreplconn->assoc_ctx.peer_ctx;
-        wrepl_in->assoc_ctx.our_ctx     = 0;
-
-        /* now we can free the wreplsrv_out_connection */
-        talloc_free(state->wreplconn);
-        state->wreplconn = NULL;
-
-        state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
+        state->subreq = wrepl_request_send(state,
+                                           state->wreplconn->service->task->event_ctx,
+                                           state->wreplconn->sock, req, NULL);
+        NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_push_notify_handler_treq,
+                                state);
+
+        state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE;
 
         return NT_STATUS_OK;
@@ -954 +915 @@
         state->ctrl.send_only           = true;
 
-        state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
-        NT_STATUS_HAVE_NO_MEMORY(state->req);
-
-        state->req->async.fn            = wreplsrv_push_notify_handler_req;
-        state->req->async.private_data  = state;
+        state->subreq = wrepl_request_send(state,
+                                           state->wreplconn->service->task->event_ctx,
+                                           state->wreplconn->sock, req, &state->ctrl);
+        NT_STATUS_HAVE_NO_MEMORY(state->subreq);
+
+        tevent_req_set_callback(state->subreq,
+                                wreplsrv_push_notify_handler_treq,
+                                state);
 
         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
@@ -1004 +968 @@
                 return NT_STATUS_INTERNAL_ERROR;
         }
-
-        return NT_STATUS_INTERNAL_ERROR;
+}
+
+static NTSTATUS wreplsrv_push_notify_wait_update(struct wreplsrv_push_notify_state *state)
+{
+        struct wreplsrv_in_connection *wrepl_in;
+        struct tstream_context *stream;
+        NTSTATUS status;
+
+        status = wrepl_request_recv(state->subreq, state, NULL);
+        TALLOC_FREE(state->subreq);
+        NT_STATUS_NOT_OK_RETURN(status);
+
+        /*
+         * now we need to convert the wrepl_socket (client connection)
+         * into a wreplsrv_in_connection (server connection), because
+         * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
+         * message is received by the peer.
+         */
+
+        status = wrepl_socket_split_stream(state->wreplconn->sock, state, &stream);
+        NT_STATUS_NOT_OK_RETURN(status);
+
+        /*
+         * now create a wreplsrv_in_connection,
+         * on which we act as server
+         *
+         * NOTE: stream will be stolen by
+         *       wreplsrv_in_connection_merge()
+         */
+        status = wreplsrv_in_connection_merge(state->io->in.partner,
+                                              state->wreplconn->assoc_ctx.peer_ctx,
+                                              &stream,
+                                              &wrepl_in);
+        NT_STATUS_NOT_OK_RETURN(status);
+
+        /* now we can free the wreplsrv_out_connection */
+        TALLOC_FREE(state->wreplconn);
+
+        state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
+        return NT_STATUS_OK;
 }
 
@@ -1012 +1014 @@
         NTSTATUS status;
 
-        status =  wrepl_request_recv(state->req, state, NULL);
+        status = wrepl_request_recv(state->subreq, state, NULL);
+        TALLOC_FREE(state->subreq);
         NT_STATUS_NOT_OK_RETURN(status);
 
@@ -1026 +1029 @@
         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
                 c->status = wreplsrv_push_notify_wait_connect(state);
+                break;
+        case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE:
+                c->status = wreplsrv_push_notify_wait_update(state);
                 break;
         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
@@ -1055 +1061 @@
 }
 
-static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
-{
-        struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private_data,
+static void wreplsrv_push_notify_handler_treq(struct tevent_req *subreq)
+{
+        struct wreplsrv_push_notify_state *state = tevent_req_callback_data(subreq,
                                                    struct wreplsrv_push_notify_state);
         wreplsrv_push_notify_handler(state);

vendor/current/source4/wrepl_server/wrepl_scavenging.c

@@ -24 +24 @@
 #include "wrepl_server/wrepl_server.h"
 #include "nbt_server/wins/winsdb.h"
-#include "ldb/include/ldb.h"
-#include "ldb/include/ldb_errors.h"
+#include <ldb.h>
+#include <ldb_errors.h>
 #include "system/time.h"
 #include "smbd/service_task.h"
 #include "lib/messaging/irpc.h"
-#include "librpc/gen_ndr/ndr_irpc.h"
+#include "librpc/gen_ndr/ndr_irpc_c.h"
 #include "librpc/gen_ndr/ndr_nbt.h"
 #include "param/param.h"
@@ -53 +53 @@
         const char *owner_filter;
         const char *filter;
-        uint32_t i;
+        unsigned int i;
         int ret;
         time_t now = time(NULL);
@@ -66 +66 @@
         struct timeval tombstone_extra_time;
         const char *local_owner = service->wins_db->local_owner;
-        bool propagate = lp_parm_bool(service->task->lp_ctx, NULL, "wreplsrv", "propagate name releases", false);
+        bool propagate = lpcfg_parm_bool(service->task->lp_ctx, NULL, "wreplsrv", "propagate name releases", false);
 
         now_timestr = ldb_timestring(tmp_mem, now);
@@ -207 +207 @@
         const char *owner_filter;
         const char *filter;
-        uint32_t i;
+        unsigned int i;
         int ret;
         time_t now = time(NULL);
@@ -313 +313 @@
 };
 
-static void verify_handler(struct irpc_request *ireq)
+static void verify_handler(struct tevent_req *subreq)
 {
-        struct verify_state *s = talloc_get_type(ireq->async.private_data,
-                                 struct verify_state);
+        struct verify_state *s =
+                tevent_req_callback_data(subreq,
+                struct verify_state);
         struct winsdb_record *rec = s->rec;
         const char *action;
@@ -337 +338 @@
          * - if an error happens do nothing
          */
-        status = irpc_call_recv(ireq);
+        status = dcerpc_nbtd_proxy_wins_challenge_r_recv(subreq, s);
+        TALLOC_FREE(subreq);
         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
                 delete_record = true;
@@ -422 +424 @@
         const char *owner_filter;
         const char *filter;
-        uint32_t i;
+        unsigned int i;
         int ret;
         time_t now = time(NULL);
         const char *now_timestr;
-        struct irpc_request *ireq;
+        struct tevent_req *subreq;
         struct verify_state *s;
-        struct server_id *nbt_servers;
-
-        nbt_servers = irpc_servers_byname(service->task->msg_ctx, tmp_mem, "nbt_server");
-        if ((nbt_servers == NULL) || (nbt_servers[0].id == 0)) {
-                return NT_STATUS_INTERNAL_ERROR;
-        }
+        struct dcerpc_binding_handle *irpc_handle;
 
         now_timestr = ldb_timestring(tmp_mem, now);
@@ -490 +487 @@
                 s->r.in.addrs[0].addr   = rec->wins_owner;
 
-                ireq = IRPC_CALL_SEND(s->msg_ctx, nbt_servers[0],
-                                      irpc, NBTD_PROXY_WINS_CHALLENGE,
-                                      &s->r, s);
-                NT_STATUS_HAVE_NO_MEMORY(ireq);
-
-                ireq->async.fn          = verify_handler;
-                ireq->async.private_data= s;
+                irpc_handle = irpc_binding_handle_by_name(s,
+                                                          service->task->msg_ctx,
+                                                          "nbt_server",
+                                                          &ndr_table_irpc);
+                if (irpc_handle == NULL) {
+                        return NT_STATUS_INTERNAL_ERROR;
+                }
+
+                subreq = dcerpc_nbtd_proxy_wins_challenge_r_send(s,
+                                                                 service->task->event_ctx,
+                                                                 irpc_handle,
+                                                                 &s->r);
+                NT_STATUS_HAVE_NO_MEMORY(subreq);
+
+                tevent_req_set_callback(subreq, verify_handler, s);
 
                 talloc_steal(service, s);

vendor/current/source4/wrepl_server/wrepl_server.c

@@ -28 +28 @@
 #include "wrepl_server/wrepl_server.h"
 #include "nbt_server/wins/winsdb.h"
-#include "ldb/include/ldb.h"
-#include "ldb/include/ldb_errors.h"
+#include <ldb.h>
+#include <ldb_errors.h>
 #include "auth/auth.h"
 #include "ldb_wrap.h"
@@ -39 +39 @@
                                                   struct loadparm_context *lp_ctx)
 {
-        return ldb_wrap_connect(mem_ctx, ev_ctx, lp_ctx, private_path(mem_ctx, 
-                                lp_ctx, lp_wins_config_url(lp_ctx)),
-                                system_session(mem_ctx, lp_ctx), NULL, 0, NULL);
+        return ldb_wrap_connect(mem_ctx, ev_ctx, lp_ctx, private_path(mem_ctx,
+                                lp_ctx, lpcfg_wins_config_url(lp_ctx)),
+                                system_session(lp_ctx), NULL, 0);
 }
 
@@ -75 +75 @@
                                      struct loadparm_context *lp_ctx)
 {
-        const char *owner = lp_parm_string(lp_ctx, NULL, "winsdb", "local_owner");
+        const char *owner = lpcfg_parm_string(lp_ctx, NULL, "winsdb", "local_owner");
 
         if (owner == NULL) {
                 struct interface *ifaces;
-                load_interfaces(service, lp_interfaces(lp_ctx), &ifaces);
+                load_interfaces(service, lpcfg_interfaces(lp_ctx), &ifaces);
                 owner = iface_n_ip(ifaces, 0);
         }
@@ -94 +94 @@
 
         /* the default renew interval is 6 days */
-        service->config.renew_interval    = lp_parm_int(lp_ctx, NULL,"wreplsrv","renew_interval", 6*24*60*60);
+        service->config.renew_interval    = lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","renew_interval", 6*24*60*60);
 
         /* the default tombstone (extinction) interval is 6 days */
-        service->config.tombstone_interval= lp_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_interval", 6*24*60*60);
+        service->config.tombstone_interval= lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_interval", 6*24*60*60);
 
         /* the default tombstone (extinction) timeout is 1 day */
-        service->config.tombstone_timeout = lp_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_timeout", 1*24*60*60);
+        service->config.tombstone_timeout = lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_timeout", 1*24*60*60);
 
         /* the default tombstone extra timeout is 3 days */
-        service->config.tombstone_extra_timeout = lp_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_extra_timeout", 3*24*60*60);
+        service->config.tombstone_extra_timeout = lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","tombstone_extra_timeout", 3*24*60*60);
 
         /* the default verify interval is 24 days */
-        service->config.verify_interval   = lp_parm_int(lp_ctx, NULL,"wreplsrv","verify_interval", 24*24*60*60);
+        service->config.verify_interval   = lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","verify_interval", 24*24*60*60);
 
         /* the default scavenging interval is 'renew_interval/2' */
-        service->config.scavenging_interval=lp_parm_int(lp_ctx, NULL,"wreplsrv","scavenging_interval",
+        service->config.scavenging_interval=lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","scavenging_interval",
                                                         service->config.renew_interval/2);
 
         /* the maximun interval to the next periodic processing event */
-        service->config.periodic_interval = lp_parm_int(lp_ctx, NULL,"wreplsrv","periodic_interval", 15);
+        service->config.periodic_interval = lpcfg_parm_int(lp_ctx, NULL,"wreplsrv","periodic_interval", 15);
 
         return NT_STATUS_OK;
@@ -140 +140 @@
         int ret;
         TALLOC_CTX *tmp_ctx;
-        int i;
+        unsigned int i;
         uint64_t new_seqnumber;
 
@@ -364 +364 @@
         TALLOC_CTX *tmp_ctx = talloc_new(service);
         struct ldb_context *ldb = service->wins_db->ldb;
-        int i;
+        unsigned int i;
         struct wreplsrv_owner *local_owner;
         const char *wins_owner;
@@ -452 +452 @@
         struct wreplsrv_service *service;
 
-        if (!lp_wins_support(task->lp_ctx)) {
+        if (!lpcfg_wins_support(task->lp_ctx)) {
                 return;
         }
@@ -486 +486 @@
 
         /* 
-         * setup listen sockets, so we can anwser requests from our partners,
+         * setup listen sockets, so we can answer requests from our partners,
          * which pull from us
          */

vendor/current/source4/wrepl_server/wrepl_server.h

@@ -19 +19 @@
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
+
+#include "system/network.h"
 
 struct wreplsrv_service;
@@ -36 +38 @@
         struct wrepl_packet rep_packet;
         bool terminate_after_send;
+
+        DATA_BLOB in;
+        DATA_BLOB out;
+        struct iovec out_iov[1];
 };
 
@@ -44 +50 @@
         struct wreplsrv_in_connection *prev,*next;
         struct stream_connection *conn;
-        struct packet_context *packet;
+        struct tstream_context *tstream;
+        struct tevent_queue *send_queue;
 
         /* our global service context */