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 3 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, see <http://www.gnu.org/licenses/>.
23 #include "lib/events/events.h"
24 #include "lib/socket/socket.h"
25 #include "smbd/service_task.h"
26 #include "smbd/service_stream.h"
27 #include "librpc/gen_ndr/winsrepl.h"
28 #include "wrepl_server/wrepl_server.h"
29 #include "nbt_server/wins/winsdb.h"
30 #include "libcli/composite/composite.h"
31 #include "libcli/wrepl/winsrepl.h"
32 #include "libcli/resolve/resolve.h"
33 #include "param/param.h"
35 enum wreplsrv_out_connect_stage {
36 WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
37 WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
38 WREPLSRV_OUT_CONNECT_STAGE_DONE
41 struct wreplsrv_out_connect_state {
42 enum wreplsrv_out_connect_stage stage;
43 struct composite_context *c;
44 struct wrepl_request *req;
45 struct composite_context *c_req;
46 struct wrepl_associate assoc_io;
47 enum winsrepl_partner_type type;
48 struct wreplsrv_out_connection *wreplconn;
51 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
52 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
54 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
58 status = wrepl_connect_recv(state->c_req);
59 NT_STATUS_NOT_OK_RETURN(status);
61 state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
62 NT_STATUS_HAVE_NO_MEMORY(state->req);
64 state->req->async.fn = wreplsrv_out_connect_handler_req;
65 state->req->async.private = state;
67 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
72 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
76 status = wrepl_associate_recv(state->req, &state->assoc_io);
77 NT_STATUS_NOT_OK_RETURN(status);
79 state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
81 if (state->type == WINSREPL_PARTNER_PUSH) {
82 state->wreplconn->partner->push.wreplconn = state->wreplconn;
83 talloc_steal(state->wreplconn->partner, state->wreplconn);
84 } else if (state->type == WINSREPL_PARTNER_PULL) {
85 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
86 talloc_steal(state->wreplconn->partner, state->wreplconn);
89 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
94 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
96 struct composite_context *c = state->c;
98 switch (state->stage) {
99 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
100 c->status = wreplsrv_out_connect_wait_socket(state);
102 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
103 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
104 c->state = COMPOSITE_STATE_DONE;
106 case WREPLSRV_OUT_CONNECT_STAGE_DONE:
107 c->status = NT_STATUS_INTERNAL_ERROR;
110 if (!NT_STATUS_IS_OK(c->status)) {
111 c->state = COMPOSITE_STATE_ERROR;
114 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
119 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
121 struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
122 struct wreplsrv_out_connect_state);
123 wreplsrv_out_connect_handler(state);
127 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
129 struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
130 struct wreplsrv_out_connect_state);
131 wreplsrv_out_connect_handler(state);
135 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
136 enum winsrepl_partner_type type,
137 struct wreplsrv_out_connection *wreplconn)
139 struct composite_context *c = NULL;
140 struct wreplsrv_service *service = partner->service;
141 struct wreplsrv_out_connect_state *state = NULL;
142 struct wreplsrv_out_connection **wreplconnp = &wreplconn;
143 bool cached_connection = false;
145 c = talloc_zero(partner, struct composite_context);
148 state = talloc_zero(c, struct wreplsrv_out_connect_state);
149 if (!state) goto failed;
153 c->state = COMPOSITE_STATE_IN_PROGRESS;
154 c->event_ctx = service->task->event_ctx;
155 c->private_data = state;
157 if (type == WINSREPL_PARTNER_PUSH) {
158 cached_connection = true;
159 wreplconn = partner->push.wreplconn;
160 wreplconnp = &partner->push.wreplconn;
161 } else if (type == WINSREPL_PARTNER_PULL) {
162 cached_connection = true;
163 wreplconn = partner->pull.wreplconn;
164 wreplconnp = &partner->pull.wreplconn;
167 /* we have a connection already, so use it */
169 if (!wreplconn->sock->dead) {
170 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
171 state->wreplconn= wreplconn;
174 } else if (!cached_connection) {
175 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
176 state->wreplconn= NULL;
180 talloc_free(wreplconn);
185 wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
186 if (!wreplconn) goto failed;
188 wreplconn->service = service;
189 wreplconn->partner = partner;
190 wreplconn->sock = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
191 if (!wreplconn->sock) goto failed;
193 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
194 state->wreplconn= wreplconn;
195 state->c_req = wrepl_connect_send(wreplconn->sock,
196 partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
198 if (!state->c_req) goto failed;
200 state->c_req->async.fn = wreplsrv_out_connect_handler_creq;
201 state->c_req->async.private_data = state;
209 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
210 struct wreplsrv_out_connection **wreplconn)
214 status = composite_wait(c);
216 if (NT_STATUS_IS_OK(status)) {
217 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
218 struct wreplsrv_out_connect_state);
219 if (state->wreplconn) {
220 *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
221 if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
223 status = NT_STATUS_INVALID_CONNECTION;
232 struct wreplsrv_pull_table_io {
234 struct wreplsrv_partner *partner;
236 struct wrepl_wins_owner *owners;
240 struct wrepl_wins_owner *owners;
244 enum wreplsrv_pull_table_stage {
245 WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
246 WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
247 WREPLSRV_PULL_TABLE_STAGE_DONE
250 struct wreplsrv_pull_table_state {
251 enum wreplsrv_pull_table_stage stage;
252 struct composite_context *c;
253 struct wrepl_request *req;
254 struct wrepl_pull_table table_io;
255 struct wreplsrv_pull_table_io *io;
256 struct composite_context *creq;
257 struct wreplsrv_out_connection *wreplconn;
260 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
262 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
266 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
267 NT_STATUS_NOT_OK_RETURN(status);
269 state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
270 state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
271 NT_STATUS_HAVE_NO_MEMORY(state->req);
273 state->req->async.fn = wreplsrv_pull_table_handler_req;
274 state->req->async.private = state;
276 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
281 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
285 status = wrepl_pull_table_recv(state->req, state, &state->table_io);
286 NT_STATUS_NOT_OK_RETURN(status);
288 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
293 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
295 struct composite_context *c = state->c;
297 switch (state->stage) {
298 case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
299 c->status = wreplsrv_pull_table_wait_connection(state);
301 case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
302 c->status = wreplsrv_pull_table_wait_table_reply(state);
303 c->state = COMPOSITE_STATE_DONE;
305 case WREPLSRV_PULL_TABLE_STAGE_DONE:
306 c->status = NT_STATUS_INTERNAL_ERROR;
309 if (!NT_STATUS_IS_OK(c->status)) {
310 c->state = COMPOSITE_STATE_ERROR;
313 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
318 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
320 struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
321 struct wreplsrv_pull_table_state);
322 wreplsrv_pull_table_handler(state);
326 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
328 struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
329 struct wreplsrv_pull_table_state);
330 wreplsrv_pull_table_handler(state);
334 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
336 struct composite_context *c = NULL;
337 struct wreplsrv_service *service = io->in.partner->service;
338 struct wreplsrv_pull_table_state *state = NULL;
340 c = talloc_zero(mem_ctx, struct composite_context);
343 state = talloc_zero(c, struct wreplsrv_pull_table_state);
344 if (!state) goto failed;
348 c->state = COMPOSITE_STATE_IN_PROGRESS;
349 c->event_ctx = service->task->event_ctx;
350 c->private_data = state;
352 if (io->in.num_owners) {
353 state->table_io.out.num_partners = io->in.num_owners;
354 state->table_io.out.partners = io->in.owners;
355 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
360 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
361 state->creq = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
362 if (!state->creq) goto failed;
364 state->creq->async.fn = wreplsrv_pull_table_handler_creq;
365 state->creq->async.private_data = state;
373 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
374 struct wreplsrv_pull_table_io *io)
378 status = composite_wait(c);
380 if (NT_STATUS_IS_OK(status)) {
381 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
382 struct wreplsrv_pull_table_state);
383 io->out.num_owners = state->table_io.out.num_partners;
384 io->out.owners = 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 = talloc_reference(mem_ctx, state->pull_io.out.names);
547 enum wreplsrv_pull_cycle_stage {
548 WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
549 WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
550 WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
551 WREPLSRV_PULL_CYCLE_STAGE_DONE
554 struct wreplsrv_pull_cycle_state {
555 enum wreplsrv_pull_cycle_stage stage;
556 struct composite_context *c;
557 struct wreplsrv_pull_cycle_io *io;
558 struct wreplsrv_pull_table_io table_io;
560 struct wreplsrv_pull_names_io names_io;
561 struct composite_context *creq;
562 struct wrepl_associate_stop assoc_stop_io;
563 struct wrepl_request *req;
566 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
567 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
569 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
571 struct wreplsrv_owner *current_owner=NULL;
572 struct wreplsrv_owner *local_owner;
574 uint64_t old_max_version = 0;
575 bool do_pull = false;
577 for (i=state->current; i < state->table_io.out.num_owners; i++) {
578 current_owner = wreplsrv_find_owner(state->io->in.partner->service,
579 state->io->in.partner->pull.table,
580 state->table_io.out.owners[i].address);
582 local_owner = wreplsrv_find_owner(state->io->in.partner->service,
583 state->io->in.partner->service->table,
584 state->table_io.out.owners[i].address);
586 * this means we are ourself the current owner,
587 * and we don't want replicate ourself
589 if (!current_owner) continue;
592 * this means we don't have any records of this owner
602 * this means the remote partner has some new records of this owner
605 if (current_owner->owner.max_version > local_owner->owner.max_version) {
607 old_max_version = local_owner->owner.max_version;
614 state->names_io.in.partner = state->io->in.partner;
615 state->names_io.in.wreplconn = state->io->in.wreplconn;
616 state->names_io.in.owner = current_owner->owner;
617 state->names_io.in.owner.min_version = old_max_version + 1;
618 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
619 NT_STATUS_HAVE_NO_MEMORY(state->creq);
621 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
622 state->creq->async.private_data = state;
624 return STATUS_MORE_ENTRIES;
630 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
634 status = wreplsrv_pull_cycle_next_owner_do_work(state);
635 if (NT_STATUS_IS_OK(status)) {
636 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
637 } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
638 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
639 status = NT_STATUS_OK;
642 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
643 state->assoc_stop_io.in.assoc_ctx = state->io->in.wreplconn->assoc_ctx.peer_ctx;
644 state->assoc_stop_io.in.reason = 0;
645 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
646 NT_STATUS_HAVE_NO_MEMORY(state->req);
648 state->req->async.fn = wreplsrv_pull_cycle_handler_req;
649 state->req->async.private = state;
651 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
657 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
662 status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
663 NT_STATUS_NOT_OK_RETURN(status);
665 /* update partner table */
666 for (i=0; i < state->table_io.out.num_owners; i++) {
667 status = wreplsrv_add_table(state->io->in.partner->service,
668 state->io->in.partner,
669 &state->io->in.partner->pull.table,
670 state->table_io.out.owners[i].address,
671 state->table_io.out.owners[i].max_version);
672 NT_STATUS_NOT_OK_RETURN(status);
675 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
676 NT_STATUS_NOT_OK_RETURN(status);
681 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
685 status = wreplsrv_apply_records(state->io->in.partner,
686 &state->names_io.in.owner,
687 state->names_io.out.num_names,
688 state->names_io.out.names);
689 NT_STATUS_NOT_OK_RETURN(status);
691 talloc_free(state->names_io.out.names);
692 ZERO_STRUCT(state->names_io);
697 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
701 status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
702 NT_STATUS_NOT_OK_RETURN(status);
705 * TODO: this should maybe an async call,
706 * because we may need some network access
707 * for conflict resolving
709 status = wreplsrv_pull_cycle_apply_records(state);
710 NT_STATUS_NOT_OK_RETURN(status);
712 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
713 NT_STATUS_NOT_OK_RETURN(status);
718 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
722 status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
723 NT_STATUS_NOT_OK_RETURN(status);
725 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
730 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
732 struct composite_context *c = state->c;
734 switch (state->stage) {
735 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
736 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
738 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
739 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
741 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
742 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
744 case WREPLSRV_PULL_CYCLE_STAGE_DONE:
745 c->status = NT_STATUS_INTERNAL_ERROR;
748 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
749 c->state = COMPOSITE_STATE_DONE;
752 if (!NT_STATUS_IS_OK(c->status)) {
753 c->state = COMPOSITE_STATE_ERROR;
756 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
761 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
763 struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
764 struct wreplsrv_pull_cycle_state);
765 wreplsrv_pull_cycle_handler(state);
769 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
771 struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
772 struct wreplsrv_pull_cycle_state);
773 wreplsrv_pull_cycle_handler(state);
777 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
779 struct composite_context *c = NULL;
780 struct wreplsrv_service *service = io->in.partner->service;
781 struct wreplsrv_pull_cycle_state *state = NULL;
783 c = talloc_zero(mem_ctx, struct composite_context);
786 state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
787 if (!state) goto failed;
791 c->state = COMPOSITE_STATE_IN_PROGRESS;
792 c->event_ctx = service->task->event_ctx;
793 c->private_data = state;
795 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
796 state->table_io.in.partner = io->in.partner;
797 state->table_io.in.num_owners = io->in.num_owners;
798 state->table_io.in.owners = io->in.owners;
799 state->creq = wreplsrv_pull_table_send(state, &state->table_io);
800 if (!state->creq) goto failed;
802 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
803 state->creq->async.private_data = state;
811 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
815 status = composite_wait(c);
821 enum wreplsrv_push_notify_stage {
822 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
823 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
824 WREPLSRV_PUSH_NOTIFY_STAGE_DONE
827 struct wreplsrv_push_notify_state {
828 enum wreplsrv_push_notify_stage stage;
829 struct composite_context *c;
830 struct wreplsrv_push_notify_io *io;
831 enum wrepl_replication_cmd command;
833 struct wrepl_send_ctrl ctrl;
834 struct wrepl_request *req;
835 struct wrepl_packet req_packet;
836 struct wrepl_packet *rep_packet;
837 struct composite_context *creq;
838 struct wreplsrv_out_connection *wreplconn;
841 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
842 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
844 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
846 struct wreplsrv_service *service = state->io->in.partner->service;
847 struct wrepl_packet *req = &state->req_packet;
848 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
849 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
850 struct wreplsrv_in_connection *wrepl_in;
852 struct socket_context *sock;
853 struct packet_context *packet;
856 /* prepare the outgoing request */
857 req->opcode = WREPL_OPCODE_BITS;
858 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
859 req->mess_type = WREPL_REPLICATION;
861 repl_out->command = state->command;
863 status = wreplsrv_fill_wrepl_table(service, state, table_out,
864 service->wins_db->local_owner, state->full_table);
865 NT_STATUS_NOT_OK_RETURN(status);
867 /* queue the request */
868 state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
869 NT_STATUS_HAVE_NO_MEMORY(state->req);
872 * now we need to convert the wrepl_socket (client connection)
873 * into a wreplsrv_in_connection (server connection), because
874 * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
875 * message is received by the peer.
878 /* steal the socket_context */
879 sock = state->wreplconn->sock->sock;
880 state->wreplconn->sock->sock = NULL;
881 talloc_steal(state, sock);
884 * steal the packet_context
885 * note the request DATA_BLOB we just send on the
886 * wrepl_socket (client connection) is still unter the
887 * packet context and will be send to the wire
889 packet = state->wreplconn->sock->packet;
890 state->wreplconn->sock->packet = NULL;
891 talloc_steal(state, packet);
894 * get the fde_flags of the old fde event,
895 * so that we can later set the same flags to the new one
897 fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
900 * free the wrepl_socket (client connection)
902 talloc_free(state->wreplconn->sock);
903 state->wreplconn->sock = NULL;
906 * now create a wreplsrv_in_connection,
907 * on which we act as server
909 * NOTE: sock and packet will be stolen by
910 * wreplsrv_in_connection_merge()
912 status = wreplsrv_in_connection_merge(state->io->in.partner,
913 sock, packet, &wrepl_in);
914 NT_STATUS_NOT_OK_RETURN(status);
916 event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
918 wrepl_in->assoc_ctx.peer_ctx = state->wreplconn->assoc_ctx.peer_ctx;
919 wrepl_in->assoc_ctx.our_ctx = 0;
921 /* now we can free the wreplsrv_out_connection */
922 talloc_free(state->wreplconn);
923 state->wreplconn = NULL;
925 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
930 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
932 struct wreplsrv_service *service = state->io->in.partner->service;
933 struct wrepl_packet *req = &state->req_packet;
934 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
935 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
938 req->opcode = WREPL_OPCODE_BITS;
939 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
940 req->mess_type = WREPL_REPLICATION;
942 repl_out->command = state->command;
944 status = wreplsrv_fill_wrepl_table(service, state, table_out,
945 service->wins_db->local_owner, state->full_table);
946 NT_STATUS_NOT_OK_RETURN(status);
948 /* we won't get a reply to a inform message */
949 state->ctrl.send_only = true;
951 state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
952 NT_STATUS_HAVE_NO_MEMORY(state->req);
954 state->req->async.fn = wreplsrv_push_notify_handler_req;
955 state->req->async.private = state;
957 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
962 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
966 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
967 NT_STATUS_NOT_OK_RETURN(status);
969 switch (state->command) {
970 case WREPL_REPL_UPDATE:
971 state->full_table = true;
972 return wreplsrv_push_notify_update(state);
973 case WREPL_REPL_UPDATE2:
974 state->full_table = false;
975 return wreplsrv_push_notify_update(state);
976 case WREPL_REPL_INFORM:
977 state->full_table = true;
978 return wreplsrv_push_notify_inform(state);
979 case WREPL_REPL_INFORM2:
980 state->full_table = false;
981 return wreplsrv_push_notify_inform(state);
983 return NT_STATUS_INTERNAL_ERROR;
986 return NT_STATUS_INTERNAL_ERROR;
989 static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
993 status = wrepl_request_recv(state->req, state, NULL);
994 NT_STATUS_NOT_OK_RETURN(status);
996 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
1000 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
1002 struct composite_context *c = state->c;
1004 switch (state->stage) {
1005 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
1006 c->status = wreplsrv_push_notify_wait_connect(state);
1008 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
1009 c->status = wreplsrv_push_notify_wait_inform(state);
1011 case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
1012 c->status = NT_STATUS_INTERNAL_ERROR;
1015 if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
1016 c->state = COMPOSITE_STATE_DONE;
1019 if (!NT_STATUS_IS_OK(c->status)) {
1020 c->state = COMPOSITE_STATE_ERROR;
1023 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
1028 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
1030 struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
1031 struct wreplsrv_push_notify_state);
1032 wreplsrv_push_notify_handler(state);
1036 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
1038 struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
1039 struct wreplsrv_push_notify_state);
1040 wreplsrv_push_notify_handler(state);
1044 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
1046 struct composite_context *c = NULL;
1047 struct wreplsrv_service *service = io->in.partner->service;
1048 struct wreplsrv_push_notify_state *state = NULL;
1049 enum winsrepl_partner_type partner_type;
1051 c = talloc_zero(mem_ctx, struct composite_context);
1052 if (!c) goto failed;
1054 state = talloc_zero(c, struct wreplsrv_push_notify_state);
1055 if (!state) goto failed;
1059 if (io->in.inform) {
1060 /* we can cache the connection in partner->push->wreplconn */
1061 partner_type = WINSREPL_PARTNER_PUSH;
1062 if (io->in.propagate) {
1063 state->command = WREPL_REPL_INFORM2;
1065 state->command = WREPL_REPL_INFORM;
1068 /* we can NOT cache the connection */
1069 partner_type = WINSREPL_PARTNER_NONE;
1070 if (io->in.propagate) {
1071 state->command = WREPL_REPL_UPDATE2;
1073 state->command = WREPL_REPL_UPDATE;
1077 c->state = COMPOSITE_STATE_IN_PROGRESS;
1078 c->event_ctx = service->task->event_ctx;
1079 c->private_data = state;
1081 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1082 state->creq = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1083 if (!state->creq) goto failed;
1085 state->creq->async.fn = wreplsrv_push_notify_handler_creq;
1086 state->creq->async.private_data = state;
1094 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1098 status = composite_wait(c);