Fix include paths to new location of libutil.
[kai/samba-autobuild/.git] / source4 / dsdb / repl / drepl_out_pull.c
1 /* 
2    Unix SMB/CIFS mplementation.
3    DSDB replication service outgoing Pull-Replication
4    
5    Copyright (C) Stefan Metzmacher 2007
6     
7    This program is free software; you can redistribute it and/or modify
8    it under the terms of the GNU General Public License as published by
9    the Free Software Foundation; either version 3 of the License, or
10    (at your option) any later version.
11    
12    This program is distributed in the hope that it will be useful,
13    but WITHOUT ANY WARRANTY; without even the implied warranty of
14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15    GNU General Public License for more details.
16    
17    You should have received a copy of the GNU General Public License
18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
19    
20 */
21
22 #include "includes.h"
23 #include "dsdb/samdb/samdb.h"
24 #include "auth/auth.h"
25 #include "smbd/service.h"
26 #include "lib/events/events.h"
27 #include "lib/messaging/irpc.h"
28 #include "dsdb/repl/drepl_service.h"
29 #include "lib/ldb/include/ldb_errors.h"
30 #include "../lib/util/dlinklist.h"
31 #include "librpc/gen_ndr/ndr_misc.h"
32 #include "librpc/gen_ndr/ndr_drsuapi.h"
33 #include "librpc/gen_ndr/ndr_drsblobs.h"
34 #include "libcli/composite/composite.h"
35
36 static WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
37                                                       struct dreplsrv_partition *p,
38                                                       struct dreplsrv_partition_source_dsa *source,
39                                                       TALLOC_CTX *mem_ctx)
40 {
41         struct dreplsrv_out_operation *op;
42
43         op = talloc_zero(mem_ctx, struct dreplsrv_out_operation);
44         W_ERROR_HAVE_NO_MEMORY(op);
45
46         op->service     = s;
47         op->source_dsa  = source;
48
49         DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
50         talloc_steal(s, op);
51         return WERR_OK;
52 }
53
54 static WERROR dreplsrv_schedule_partition_pull(struct dreplsrv_service *s,
55                                                struct dreplsrv_partition *p,
56                                                TALLOC_CTX *mem_ctx)
57 {
58         WERROR status;
59         struct dreplsrv_partition_source_dsa *cur;
60
61         for (cur = p->sources; cur; cur = cur->next) {
62                 status = dreplsrv_schedule_partition_pull_source(s, p, cur, mem_ctx);
63                 W_ERROR_NOT_OK_RETURN(status);
64         }
65
66         return WERR_OK;
67 }
68
69 WERROR dreplsrv_schedule_pull_replication(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
70 {
71         WERROR status;
72         struct dreplsrv_partition *p;
73
74         for (p = s->partitions; p; p = p->next) {
75                 status = dreplsrv_schedule_partition_pull(s, p, mem_ctx);
76                 W_ERROR_NOT_OK_RETURN(status);
77         }
78
79         return WERR_OK;
80 }
81
82 static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
83 {
84         struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
85         struct dreplsrv_service *s = op->service;
86         time_t t;
87         NTTIME now;
88
89         t = time(NULL);
90         unix_to_nt_time(&now, t);
91
92         rf->result_last_attempt = dreplsrv_op_pull_source_recv(op->creq);
93         if (W_ERROR_IS_OK(rf->result_last_attempt)) {
94                 rf->consecutive_sync_failures   = 0;
95                 rf->last_success                = now;
96                 DEBUG(2,("dreplsrv_op_pull_source(%s)\n",
97                         win_errstr(rf->result_last_attempt)));
98                 goto done;
99         }
100
101         rf->consecutive_sync_failures++;
102
103         DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
104                 win_errstr(rf->result_last_attempt),
105                 nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
106                 rf->consecutive_sync_failures));
107
108 done:
109         talloc_free(op);
110         s->ops.current = NULL;
111         dreplsrv_run_pending_ops(s);
112 }
113
114 static void dreplsrv_pending_op_callback_creq(struct composite_context *creq)
115 {
116         struct dreplsrv_out_operation *op = talloc_get_type(creq->async.private_data,
117                                                            struct dreplsrv_out_operation);
118         dreplsrv_pending_op_callback(op);
119 }
120
121 void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
122 {
123         struct dreplsrv_out_operation *op;
124         time_t t;
125         NTTIME now;
126
127         if (s->ops.current) {
128                 /* if there's still one running, we're done */
129                 return;
130         }
131
132         if (!s->ops.pending) {
133                 /* if there're no pending operations, we're done */
134                 return;
135         }
136
137         t = time(NULL);
138         unix_to_nt_time(&now, t);
139
140         op = s->ops.pending;
141         s->ops.current = op;
142         DLIST_REMOVE(s->ops.pending, op);
143
144         op->source_dsa->repsFrom1->last_attempt = now;
145
146         op->creq = dreplsrv_op_pull_source_send(op);
147         if (!op->creq) {
148                 dreplsrv_pending_op_callback(op);
149                 return;
150         }
151
152         op->creq->async.fn              = dreplsrv_pending_op_callback_creq;
153         op->creq->async.private_data    = op;
154 }