2 Unix SMB/CIFS implementation.
4 WINS Replication server
6 Copyright (C) Stefan Metzmacher 2005
8 This program is free software; you can redistribute it and/or modify
9 it under the terms of the GNU General Public License as published by
10 the Free Software Foundation; either version 2 of the License, or
11 (at your option) any later version.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; if not, write to the Free Software
20 Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24 #include "lib/events/events.h"
25 #include "lib/socket/socket.h"
26 #include "smbd/service_task.h"
27 #include "smbd/service_stream.h"
28 #include "librpc/gen_ndr/winsrepl.h"
29 #include "wrepl_server/wrepl_server.h"
30 #include "nbt_server/wins/winsdb.h"
31 #include "libcli/composite/composite.h"
32 #include "libcli/wrepl/winsrepl.h"
34 enum wreplsrv_out_connect_stage {
35 WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
36 WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
37 WREPLSRV_OUT_CONNECT_STAGE_DONE
40 struct wreplsrv_out_connect_state {
41 enum wreplsrv_out_connect_stage stage;
42 struct composite_context *c;
43 struct wrepl_request *req;
44 struct composite_context *c_req;
45 struct wrepl_associate assoc_io;
46 enum winsrepl_partner_type type;
47 struct wreplsrv_out_connection *wreplconn;
50 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
51 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
53 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
57 status = wrepl_connect_recv(state->c_req);
58 NT_STATUS_NOT_OK_RETURN(status);
60 state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
61 NT_STATUS_HAVE_NO_MEMORY(state->req);
63 state->req->async.fn = wreplsrv_out_connect_handler_req;
64 state->req->async.private = state;
66 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
71 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
75 status = wrepl_associate_recv(state->req, &state->assoc_io);
76 NT_STATUS_NOT_OK_RETURN(status);
78 state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
80 if (state->type == WINSREPL_PARTNER_PUSH) {
81 state->wreplconn->partner->push.wreplconn = state->wreplconn;
82 talloc_steal(state->wreplconn->partner, state->wreplconn);
83 } else if (state->type == WINSREPL_PARTNER_PULL) {
84 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
85 talloc_steal(state->wreplconn->partner, state->wreplconn);
88 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
93 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
95 struct composite_context *c = state->c;
97 switch (state->stage) {
98 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
99 c->status = wreplsrv_out_connect_wait_socket(state);
101 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
102 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
103 c->state = COMPOSITE_STATE_DONE;
105 case WREPLSRV_OUT_CONNECT_STAGE_DONE:
106 c->status = NT_STATUS_INTERNAL_ERROR;
109 if (!NT_STATUS_IS_OK(c->status)) {
110 c->state = COMPOSITE_STATE_ERROR;
113 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
118 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
120 struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
121 struct wreplsrv_out_connect_state);
122 wreplsrv_out_connect_handler(state);
126 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
128 struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
129 struct wreplsrv_out_connect_state);
130 wreplsrv_out_connect_handler(state);
134 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
135 enum winsrepl_partner_type type,
136 struct wreplsrv_out_connection *wreplconn)
138 struct composite_context *c = NULL;
139 struct wreplsrv_service *service = partner->service;
140 struct wreplsrv_out_connect_state *state = NULL;
141 struct wreplsrv_out_connection **wreplconnp = &wreplconn;
142 BOOL cached_connection = False;
144 c = talloc_zero(partner, struct composite_context);
147 state = talloc_zero(c, struct wreplsrv_out_connect_state);
148 if (!state) goto failed;
152 c->state = COMPOSITE_STATE_IN_PROGRESS;
153 c->event_ctx = service->task->event_ctx;
154 c->private_data = state;
156 if (type == WINSREPL_PARTNER_PUSH) {
157 cached_connection = True;
158 wreplconn = partner->push.wreplconn;
159 wreplconnp = &partner->push.wreplconn;
160 } else if (type == WINSREPL_PARTNER_PULL) {
161 cached_connection = True;
162 wreplconn = partner->pull.wreplconn;
163 wreplconnp = &partner->pull.wreplconn;
166 /* we have a connection already, so use it */
168 if (!wreplconn->sock->dead) {
169 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
170 state->wreplconn= wreplconn;
173 } else if (!cached_connection) {
174 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
175 state->wreplconn= NULL;
179 talloc_free(wreplconn);
184 wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
185 if (!wreplconn) goto failed;
187 wreplconn->service = service;
188 wreplconn->partner = partner;
189 wreplconn->sock = wrepl_socket_init(wreplconn, service->task->event_ctx);
190 if (!wreplconn->sock) goto failed;
192 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
193 state->wreplconn= wreplconn;
194 state->c_req = wrepl_connect_send(wreplconn->sock,
195 partner->our_address,
197 if (!state->c_req) goto failed;
199 state->c_req->async.fn = wreplsrv_out_connect_handler_creq;
200 state->c_req->async.private_data = state;
208 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
209 struct wreplsrv_out_connection **wreplconn)
213 status = composite_wait(c);
215 if (NT_STATUS_IS_OK(status)) {
216 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
217 struct wreplsrv_out_connect_state);
218 if (state->wreplconn) {
219 *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
220 if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
222 status = NT_STATUS_INVALID_CONNECTION;
231 struct wreplsrv_pull_table_io {
233 struct wreplsrv_partner *partner;
235 struct wrepl_wins_owner *owners;
239 struct wrepl_wins_owner *owners;
243 enum wreplsrv_pull_table_stage {
244 WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
245 WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
246 WREPLSRV_PULL_TABLE_STAGE_DONE
249 struct wreplsrv_pull_table_state {
250 enum wreplsrv_pull_table_stage stage;
251 struct composite_context *c;
252 struct wrepl_request *req;
253 struct wrepl_pull_table table_io;
254 struct wreplsrv_pull_table_io *io;
255 struct composite_context *creq;
256 struct wreplsrv_out_connection *wreplconn;
259 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
261 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
265 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
266 NT_STATUS_NOT_OK_RETURN(status);
268 state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
269 state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
270 NT_STATUS_HAVE_NO_MEMORY(state->req);
272 state->req->async.fn = wreplsrv_pull_table_handler_req;
273 state->req->async.private = state;
275 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
280 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
284 status = wrepl_pull_table_recv(state->req, state, &state->table_io);
285 NT_STATUS_NOT_OK_RETURN(status);
287 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
292 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
294 struct composite_context *c = state->c;
296 switch (state->stage) {
297 case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
298 c->status = wreplsrv_pull_table_wait_connection(state);
300 case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
301 c->status = wreplsrv_pull_table_wait_table_reply(state);
302 c->state = COMPOSITE_STATE_DONE;
304 case WREPLSRV_PULL_TABLE_STAGE_DONE:
305 c->status = NT_STATUS_INTERNAL_ERROR;
308 if (!NT_STATUS_IS_OK(c->status)) {
309 c->state = COMPOSITE_STATE_ERROR;
312 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
317 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
319 struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
320 struct wreplsrv_pull_table_state);
321 wreplsrv_pull_table_handler(state);
325 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
327 struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
328 struct wreplsrv_pull_table_state);
329 wreplsrv_pull_table_handler(state);
333 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
335 struct composite_context *c = NULL;
336 struct wreplsrv_service *service = io->in.partner->service;
337 struct wreplsrv_pull_table_state *state = NULL;
339 c = talloc_zero(mem_ctx, struct composite_context);
342 state = talloc_zero(c, struct wreplsrv_pull_table_state);
343 if (!state) goto failed;
347 c->state = COMPOSITE_STATE_IN_PROGRESS;
348 c->event_ctx = service->task->event_ctx;
349 c->private_data = state;
351 if (io->in.num_owners) {
352 state->table_io.out.num_partners = io->in.num_owners;
353 state->table_io.out.partners = io->in.owners;
354 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
359 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
360 state->creq = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
361 if (!state->creq) goto failed;
363 state->creq->async.fn = wreplsrv_pull_table_handler_creq;
364 state->creq->async.private_data = state;
372 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
373 struct wreplsrv_pull_table_io *io)
377 status = composite_wait(c);
379 if (NT_STATUS_IS_OK(status)) {
380 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
381 struct wreplsrv_pull_table_state);
382 io->out.num_owners = state->table_io.out.num_partners;
383 io->out.owners = state->table_io.out.partners;
384 talloc_reference(mem_ctx, state->table_io.out.partners);
391 struct wreplsrv_pull_names_io {
393 struct wreplsrv_partner *partner;
394 struct wreplsrv_out_connection *wreplconn;
395 struct wrepl_wins_owner owner;
399 struct wrepl_name *names;
403 enum wreplsrv_pull_names_stage {
404 WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
405 WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
406 WREPLSRV_PULL_NAMES_STAGE_DONE
409 struct wreplsrv_pull_names_state {
410 enum wreplsrv_pull_names_stage stage;
411 struct composite_context *c;
412 struct wrepl_request *req;
413 struct wrepl_pull_names pull_io;
414 struct wreplsrv_pull_names_io *io;
415 struct composite_context *creq;
416 struct wreplsrv_out_connection *wreplconn;
419 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
421 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
425 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
426 NT_STATUS_NOT_OK_RETURN(status);
428 state->pull_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
429 state->pull_io.in.partner = state->io->in.owner;
430 state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
431 NT_STATUS_HAVE_NO_MEMORY(state->req);
433 state->req->async.fn = wreplsrv_pull_names_handler_req;
434 state->req->async.private = state;
436 state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
441 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
445 status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
446 NT_STATUS_NOT_OK_RETURN(status);
448 state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
453 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
455 struct composite_context *c = state->c;
457 switch (state->stage) {
458 case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
459 c->status = wreplsrv_pull_names_wait_connection(state);
461 case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
462 c->status = wreplsrv_pull_names_wait_send_reply(state);
463 c->state = COMPOSITE_STATE_DONE;
465 case WREPLSRV_PULL_NAMES_STAGE_DONE:
466 c->status = NT_STATUS_INTERNAL_ERROR;
469 if (!NT_STATUS_IS_OK(c->status)) {
470 c->state = COMPOSITE_STATE_ERROR;
473 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
478 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
480 struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
481 struct wreplsrv_pull_names_state);
482 wreplsrv_pull_names_handler(state);
486 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
488 struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
489 struct wreplsrv_pull_names_state);
490 wreplsrv_pull_names_handler(state);
494 static struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
496 struct composite_context *c = NULL;
497 struct wreplsrv_service *service = io->in.partner->service;
498 struct wreplsrv_pull_names_state *state = NULL;
499 enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
501 if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
503 c = talloc_zero(mem_ctx, struct composite_context);
506 state = talloc_zero(c, struct wreplsrv_pull_names_state);
507 if (!state) goto failed;
511 c->state = COMPOSITE_STATE_IN_PROGRESS;
512 c->event_ctx = service->task->event_ctx;
513 c->private_data = state;
515 state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
516 state->creq = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
517 if (!state->creq) goto failed;
519 state->creq->async.fn = wreplsrv_pull_names_handler_creq;
520 state->creq->async.private_data = state;
528 static NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
529 struct wreplsrv_pull_names_io *io)
533 status = composite_wait(c);
535 if (NT_STATUS_IS_OK(status)) {
536 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
537 struct wreplsrv_pull_names_state);
538 io->out.num_names = state->pull_io.out.num_names;
539 io->out.names = state->pull_io.out.names;
540 talloc_reference(mem_ctx, state->pull_io.out.names);
548 enum wreplsrv_pull_cycle_stage {
549 WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
550 WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
551 WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
552 WREPLSRV_PULL_CYCLE_STAGE_DONE
555 struct wreplsrv_pull_cycle_state {
556 enum wreplsrv_pull_cycle_stage stage;
557 struct composite_context *c;
558 struct wreplsrv_pull_cycle_io *io;
559 struct wreplsrv_pull_table_io table_io;
561 struct wreplsrv_pull_names_io names_io;
562 struct composite_context *creq;
563 struct wrepl_associate_stop assoc_stop_io;
564 struct wrepl_request *req;
567 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
568 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
570 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
572 struct wreplsrv_owner *current_owner=NULL;
573 struct wreplsrv_owner *local_owner;
575 uint64_t old_max_version = 0;
576 BOOL do_pull = False;
578 for (i=state->current; i < state->table_io.out.num_owners; i++) {
579 current_owner = wreplsrv_find_owner(state->io->in.partner->service,
580 state->io->in.partner->pull.table,
581 state->table_io.out.owners[i].address);
583 local_owner = wreplsrv_find_owner(state->io->in.partner->service,
584 state->io->in.partner->service->table,
585 state->table_io.out.owners[i].address);
587 * this means we are ourself the current owner,
588 * and we don't want replicate ourself
590 if (!current_owner) continue;
593 * this means we don't have any records of this owner
603 * this means the remote partner has some new records of this owner
606 if (current_owner->owner.max_version > local_owner->owner.max_version) {
608 old_max_version = local_owner->owner.max_version;
615 state->names_io.in.partner = state->io->in.partner;
616 state->names_io.in.wreplconn = state->io->in.wreplconn;
617 state->names_io.in.owner = current_owner->owner;
618 state->names_io.in.owner.min_version = old_max_version + 1;
619 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
620 NT_STATUS_HAVE_NO_MEMORY(state->creq);
622 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
623 state->creq->async.private_data = state;
625 return STATUS_MORE_ENTRIES;
631 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
635 status = wreplsrv_pull_cycle_next_owner_do_work(state);
636 if (NT_STATUS_IS_OK(status)) {
637 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
638 } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
639 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
640 status = NT_STATUS_OK;
643 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
644 state->assoc_stop_io.in.assoc_ctx = state->io->in.wreplconn->assoc_ctx.peer_ctx;
645 state->assoc_stop_io.in.reason = 0;
646 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
647 NT_STATUS_HAVE_NO_MEMORY(state->req);
649 state->req->async.fn = wreplsrv_pull_cycle_handler_req;
650 state->req->async.private = state;
652 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
658 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
663 status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
664 NT_STATUS_NOT_OK_RETURN(status);
666 /* update partner table */
667 for (i=0; i < state->table_io.out.num_owners; i++) {
668 status = wreplsrv_add_table(state->io->in.partner->service,
669 state->io->in.partner,
670 &state->io->in.partner->pull.table,
671 state->table_io.out.owners[i].address,
672 state->table_io.out.owners[i].max_version);
673 NT_STATUS_NOT_OK_RETURN(status);
676 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
677 NT_STATUS_NOT_OK_RETURN(status);
682 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
686 status = wreplsrv_apply_records(state->io->in.partner,
687 &state->names_io.in.owner,
688 state->names_io.out.num_names,
689 state->names_io.out.names);
690 NT_STATUS_NOT_OK_RETURN(status);
692 talloc_free(state->names_io.out.names);
693 ZERO_STRUCT(state->names_io);
698 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
702 status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
703 NT_STATUS_NOT_OK_RETURN(status);
706 * TODO: this should maybe an async call,
707 * because we may need some network access
708 * for conflict resolving
710 status = wreplsrv_pull_cycle_apply_records(state);
711 NT_STATUS_NOT_OK_RETURN(status);
713 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
714 NT_STATUS_NOT_OK_RETURN(status);
719 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
723 status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
724 NT_STATUS_NOT_OK_RETURN(status);
726 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
731 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
733 struct composite_context *c = state->c;
735 switch (state->stage) {
736 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
737 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
739 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
740 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
742 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
743 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
745 case WREPLSRV_PULL_CYCLE_STAGE_DONE:
746 c->status = NT_STATUS_INTERNAL_ERROR;
749 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
750 c->state = COMPOSITE_STATE_DONE;
753 if (!NT_STATUS_IS_OK(c->status)) {
754 c->state = COMPOSITE_STATE_ERROR;
757 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
762 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
764 struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
765 struct wreplsrv_pull_cycle_state);
766 wreplsrv_pull_cycle_handler(state);
770 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
772 struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
773 struct wreplsrv_pull_cycle_state);
774 wreplsrv_pull_cycle_handler(state);
778 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
780 struct composite_context *c = NULL;
781 struct wreplsrv_service *service = io->in.partner->service;
782 struct wreplsrv_pull_cycle_state *state = NULL;
784 c = talloc_zero(mem_ctx, struct composite_context);
787 state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
788 if (!state) goto failed;
792 c->state = COMPOSITE_STATE_IN_PROGRESS;
793 c->event_ctx = service->task->event_ctx;
794 c->private_data = state;
796 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
797 state->table_io.in.partner = io->in.partner;
798 state->table_io.in.num_owners = io->in.num_owners;
799 state->table_io.in.owners = io->in.owners;
800 state->creq = wreplsrv_pull_table_send(state, &state->table_io);
801 if (!state->creq) goto failed;
803 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
804 state->creq->async.private_data = state;
812 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
816 status = composite_wait(c);
822 enum wreplsrv_push_notify_stage {
823 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
824 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
825 WREPLSRV_PUSH_NOTIFY_STAGE_DONE
828 struct wreplsrv_push_notify_state {
829 enum wreplsrv_push_notify_stage stage;
830 struct composite_context *c;
831 struct wreplsrv_push_notify_io *io;
832 enum wrepl_replication_cmd command;
834 struct wrepl_send_ctrl ctrl;
835 struct wrepl_request *req;
836 struct wrepl_packet req_packet;
837 struct wrepl_packet *rep_packet;
838 struct composite_context *creq;
839 struct wreplsrv_out_connection *wreplconn;
842 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
843 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
845 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
847 struct wreplsrv_service *service = state->io->in.partner->service;
848 struct wrepl_packet *req = &state->req_packet;
849 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
850 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
851 struct wreplsrv_in_connection *wrepl_in;
853 struct socket_context *sock;
854 struct packet_context *packet;
857 /* prepare the outgoing request */
858 req->opcode = WREPL_OPCODE_BITS;
859 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
860 req->mess_type = WREPL_REPLICATION;
862 repl_out->command = state->command;
864 status = wreplsrv_fill_wrepl_table(service, state, table_out,
865 service->wins_db->local_owner, state->full_table);
866 NT_STATUS_NOT_OK_RETURN(status);
868 /* queue the request */
869 state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
870 NT_STATUS_HAVE_NO_MEMORY(state->req);
873 * now we need to convert the wrepl_socket (client connection)
874 * into a wreplsrv_in_connection (server connection), because
875 * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
876 * message is received by the peer.
879 /* steal the socket_context */
880 sock = state->wreplconn->sock->sock;
881 state->wreplconn->sock->sock = NULL;
882 talloc_steal(state, sock);
885 * steal the packet_context
886 * note the request DATA_BLOB we just send on the
887 * wrepl_socket (client connection) is still unter the
888 * packet context and will be send to the wire
890 packet = state->wreplconn->sock->packet;
891 state->wreplconn->sock->packet = NULL;
892 talloc_steal(state, packet);
895 * get the fde_flags of the old fde event,
896 * so that we can later set the same flags to the new one
898 fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
901 * free the wrepl_socket (client connection)
903 talloc_free(state->wreplconn->sock);
904 state->wreplconn->sock = NULL;
907 * now create a wreplsrv_in_connection,
908 * on which we act as server
910 * NOTE: sock and packet will be stolen by
911 * wreplsrv_in_connection_merge()
913 status = wreplsrv_in_connection_merge(state->io->in.partner,
914 sock, packet, &wrepl_in);
915 NT_STATUS_NOT_OK_RETURN(status);
917 event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
919 wrepl_in->assoc_ctx.peer_ctx = state->wreplconn->assoc_ctx.peer_ctx;
920 wrepl_in->assoc_ctx.our_ctx = 0;
922 /* now we can free the wreplsrv_out_connection */
923 talloc_free(state->wreplconn);
924 state->wreplconn = NULL;
926 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
931 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
933 struct wreplsrv_service *service = state->io->in.partner->service;
934 struct wrepl_packet *req = &state->req_packet;
935 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
936 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
939 req->opcode = WREPL_OPCODE_BITS;
940 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
941 req->mess_type = WREPL_REPLICATION;
943 repl_out->command = state->command;
945 status = wreplsrv_fill_wrepl_table(service, state, table_out,
946 service->wins_db->local_owner, state->full_table);
947 NT_STATUS_NOT_OK_RETURN(status);
949 /* we won't get a reply to a inform message */
950 state->ctrl.send_only = True;
952 state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
953 NT_STATUS_HAVE_NO_MEMORY(state->req);
955 state->req->async.fn = wreplsrv_push_notify_handler_req;
956 state->req->async.private = state;
958 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
963 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
967 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
968 NT_STATUS_NOT_OK_RETURN(status);
970 switch (state->command) {
971 case WREPL_REPL_UPDATE:
972 state->full_table = True;
973 return wreplsrv_push_notify_update(state);
974 case WREPL_REPL_UPDATE2:
975 state->full_table = False;
976 return wreplsrv_push_notify_update(state);
977 case WREPL_REPL_INFORM:
978 state->full_table = True;
979 return wreplsrv_push_notify_inform(state);
980 case WREPL_REPL_INFORM2:
981 state->full_table = False;
982 return wreplsrv_push_notify_inform(state);
984 return NT_STATUS_INTERNAL_ERROR;
987 return NT_STATUS_INTERNAL_ERROR;
990 static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
994 status = wrepl_request_recv(state->req, state, NULL);
995 NT_STATUS_NOT_OK_RETURN(status);
997 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
1001 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
1003 struct composite_context *c = state->c;
1005 switch (state->stage) {
1006 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
1007 c->status = wreplsrv_push_notify_wait_connect(state);
1009 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
1010 c->status = wreplsrv_push_notify_wait_inform(state);
1012 case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
1013 c->status = NT_STATUS_INTERNAL_ERROR;
1016 if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
1017 c->state = COMPOSITE_STATE_DONE;
1020 if (!NT_STATUS_IS_OK(c->status)) {
1021 c->state = COMPOSITE_STATE_ERROR;
1024 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
1029 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
1031 struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
1032 struct wreplsrv_push_notify_state);
1033 wreplsrv_push_notify_handler(state);
1037 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
1039 struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
1040 struct wreplsrv_push_notify_state);
1041 wreplsrv_push_notify_handler(state);
1045 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
1047 struct composite_context *c = NULL;
1048 struct wreplsrv_service *service = io->in.partner->service;
1049 struct wreplsrv_push_notify_state *state = NULL;
1050 enum winsrepl_partner_type partner_type;
1052 c = talloc_zero(mem_ctx, struct composite_context);
1053 if (!c) goto failed;
1055 state = talloc_zero(c, struct wreplsrv_push_notify_state);
1056 if (!state) goto failed;
1060 if (io->in.inform) {
1061 /* we can cache the connection in partner->push->wreplconn */
1062 partner_type = WINSREPL_PARTNER_PUSH;
1063 if (io->in.propagate) {
1064 state->command = WREPL_REPL_INFORM2;
1066 state->command = WREPL_REPL_INFORM;
1069 /* we can NOT cache the connection */
1070 partner_type = WINSREPL_PARTNER_NONE;
1071 if (io->in.propagate) {
1072 state->command = WREPL_REPL_UPDATE2;
1074 state->command = WREPL_REPL_UPDATE;
1078 c->state = COMPOSITE_STATE_IN_PROGRESS;
1079 c->event_ctx = service->task->event_ctx;
1080 c->private_data = state;
1082 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1083 state->creq = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1084 if (!state->creq) goto failed;
1086 state->creq->async.fn = wreplsrv_push_notify_handler_creq;
1087 state->creq->async.private_data = state;
1095 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1099 status = composite_wait(c);