• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /*
3    drbd.c
4 
5    This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 
7    Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
8    Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
9    Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 
11    Thanks to Carter Burden, Bart Grantham and Gennadiy Nerubayev
12    from Logicworks, Inc. for making SDP replication support possible.
13 
14 
15  */
16 
17 #define pr_fmt(fmt)	KBUILD_MODNAME ": " fmt
18 
19 #include <linux/module.h>
20 #include <linux/jiffies.h>
21 #include <linux/drbd.h>
22 #include <linux/uaccess.h>
23 #include <asm/types.h>
24 #include <net/sock.h>
25 #include <linux/ctype.h>
26 #include <linux/mutex.h>
27 #include <linux/fs.h>
28 #include <linux/file.h>
29 #include <linux/proc_fs.h>
30 #include <linux/init.h>
31 #include <linux/mm.h>
32 #include <linux/memcontrol.h>
33 #include <linux/mm_inline.h>
34 #include <linux/slab.h>
35 #include <linux/random.h>
36 #include <linux/reboot.h>
37 #include <linux/notifier.h>
38 #include <linux/kthread.h>
39 #include <linux/workqueue.h>
40 #define __KERNEL_SYSCALLS__
41 #include <linux/unistd.h>
42 #include <linux/vmalloc.h>
43 #include <linux/sched/signal.h>
44 
45 #include <linux/drbd_limits.h>
46 #include "drbd_int.h"
47 #include "drbd_protocol.h"
48 #include "drbd_req.h" /* only for _req_mod in tl_release and tl_clear */
49 #include "drbd_vli.h"
50 #include "drbd_debugfs.h"
51 
52 static DEFINE_MUTEX(drbd_main_mutex);
53 static int drbd_open(struct block_device *bdev, fmode_t mode);
54 static void drbd_release(struct gendisk *gd, fmode_t mode);
55 static void md_sync_timer_fn(struct timer_list *t);
56 static int w_bitmap_io(struct drbd_work *w, int unused);
57 
58 MODULE_AUTHOR("Philipp Reisner <phil@linbit.com>, "
59 	      "Lars Ellenberg <lars@linbit.com>");
60 MODULE_DESCRIPTION("drbd - Distributed Replicated Block Device v" REL_VERSION);
61 MODULE_VERSION(REL_VERSION);
62 MODULE_LICENSE("GPL");
63 MODULE_PARM_DESC(minor_count, "Approximate number of drbd devices ("
64 		 __stringify(DRBD_MINOR_COUNT_MIN) "-" __stringify(DRBD_MINOR_COUNT_MAX) ")");
65 MODULE_ALIAS_BLOCKDEV_MAJOR(DRBD_MAJOR);
66 
67 #include <linux/moduleparam.h>
68 /* thanks to these macros, if compiled into the kernel (not-module),
69  * these become boot parameters (e.g., drbd.minor_count) */
70 
71 #ifdef CONFIG_DRBD_FAULT_INJECTION
72 int drbd_enable_faults;
73 int drbd_fault_rate;
74 static int drbd_fault_count;
75 static int drbd_fault_devs;
76 /* bitmap of enabled faults */
77 module_param_named(enable_faults, drbd_enable_faults, int, 0664);
78 /* fault rate % value - applies to all enabled faults */
79 module_param_named(fault_rate, drbd_fault_rate, int, 0664);
80 /* count of faults inserted */
81 module_param_named(fault_count, drbd_fault_count, int, 0664);
82 /* bitmap of devices to insert faults on */
83 module_param_named(fault_devs, drbd_fault_devs, int, 0644);
84 #endif
85 
86 /* module parameters we can keep static */
87 static bool drbd_allow_oos; /* allow_open_on_secondary */
88 static bool drbd_disable_sendpage;
89 MODULE_PARM_DESC(allow_oos, "DONT USE!");
90 module_param_named(allow_oos, drbd_allow_oos, bool, 0);
91 module_param_named(disable_sendpage, drbd_disable_sendpage, bool, 0644);
92 
93 /* module parameters we share */
94 int drbd_proc_details; /* Detail level in proc drbd*/
95 module_param_named(proc_details, drbd_proc_details, int, 0644);
96 /* module parameters shared with defaults */
97 unsigned int drbd_minor_count = DRBD_MINOR_COUNT_DEF;
98 /* Module parameter for setting the user mode helper program
99  * to run. Default is /sbin/drbdadm */
100 char drbd_usermode_helper[80] = "/sbin/drbdadm";
101 module_param_named(minor_count, drbd_minor_count, uint, 0444);
102 module_param_string(usermode_helper, drbd_usermode_helper, sizeof(drbd_usermode_helper), 0644);
103 
104 /* in 2.6.x, our device mapping and config info contains our virtual gendisks
105  * as member "struct gendisk *vdisk;"
106  */
107 struct idr drbd_devices;
108 struct list_head drbd_resources;
109 struct mutex resources_mutex;
110 
111 struct kmem_cache *drbd_request_cache;
112 struct kmem_cache *drbd_ee_cache;	/* peer requests */
113 struct kmem_cache *drbd_bm_ext_cache;	/* bitmap extents */
114 struct kmem_cache *drbd_al_ext_cache;	/* activity log extents */
115 mempool_t drbd_request_mempool;
116 mempool_t drbd_ee_mempool;
117 mempool_t drbd_md_io_page_pool;
118 struct bio_set drbd_md_io_bio_set;
119 struct bio_set drbd_io_bio_set;
120 
121 /* I do not use a standard mempool, because:
122    1) I want to hand out the pre-allocated objects first.
123    2) I want to be able to interrupt sleeping allocation with a signal.
124    Note: This is a single linked list, the next pointer is the private
125 	 member of struct page.
126  */
127 struct page *drbd_pp_pool;
128 DEFINE_SPINLOCK(drbd_pp_lock);
129 int          drbd_pp_vacant;
130 wait_queue_head_t drbd_pp_wait;
131 
132 DEFINE_RATELIMIT_STATE(drbd_ratelimit_state, 5 * HZ, 5);
133 
134 static const struct block_device_operations drbd_ops = {
135 	.owner		= THIS_MODULE,
136 	.submit_bio	= drbd_submit_bio,
137 	.open		= drbd_open,
138 	.release	= drbd_release,
139 };
140 
141 #ifdef __CHECKER__
142 /* When checking with sparse, and this is an inline function, sparse will
143    give tons of false positives. When this is a real functions sparse works.
144  */
_get_ldev_if_state(struct drbd_device * device,enum drbd_disk_state mins)145 int _get_ldev_if_state(struct drbd_device *device, enum drbd_disk_state mins)
146 {
147 	int io_allowed;
148 
149 	atomic_inc(&device->local_cnt);
150 	io_allowed = (device->state.disk >= mins);
151 	if (!io_allowed) {
152 		if (atomic_dec_and_test(&device->local_cnt))
153 			wake_up(&device->misc_wait);
154 	}
155 	return io_allowed;
156 }
157 
158 #endif
159 
160 /**
161  * tl_release() - mark as BARRIER_ACKED all requests in the corresponding transfer log epoch
162  * @connection:	DRBD connection.
163  * @barrier_nr:	Expected identifier of the DRBD write barrier packet.
164  * @set_size:	Expected number of requests before that barrier.
165  *
166  * In case the passed barrier_nr or set_size does not match the oldest
167  * epoch of not yet barrier-acked requests, this function will cause a
168  * termination of the connection.
169  */
tl_release(struct drbd_connection * connection,unsigned int barrier_nr,unsigned int set_size)170 void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
171 		unsigned int set_size)
172 {
173 	struct drbd_request *r;
174 	struct drbd_request *req = NULL, *tmp = NULL;
175 	int expect_epoch = 0;
176 	int expect_size = 0;
177 
178 	spin_lock_irq(&connection->resource->req_lock);
179 
180 	/* find oldest not yet barrier-acked write request,
181 	 * count writes in its epoch. */
182 	list_for_each_entry(r, &connection->transfer_log, tl_requests) {
183 		const unsigned s = r->rq_state;
184 		if (!req) {
185 			if (!(s & RQ_WRITE))
186 				continue;
187 			if (!(s & RQ_NET_MASK))
188 				continue;
189 			if (s & RQ_NET_DONE)
190 				continue;
191 			req = r;
192 			expect_epoch = req->epoch;
193 			expect_size ++;
194 		} else {
195 			if (r->epoch != expect_epoch)
196 				break;
197 			if (!(s & RQ_WRITE))
198 				continue;
199 			/* if (s & RQ_DONE): not expected */
200 			/* if (!(s & RQ_NET_MASK)): not expected */
201 			expect_size++;
202 		}
203 	}
204 
205 	/* first some paranoia code */
206 	if (req == NULL) {
207 		drbd_err(connection, "BAD! BarrierAck #%u received, but no epoch in tl!?\n",
208 			 barrier_nr);
209 		goto bail;
210 	}
211 	if (expect_epoch != barrier_nr) {
212 		drbd_err(connection, "BAD! BarrierAck #%u received, expected #%u!\n",
213 			 barrier_nr, expect_epoch);
214 		goto bail;
215 	}
216 
217 	if (expect_size != set_size) {
218 		drbd_err(connection, "BAD! BarrierAck #%u received with n_writes=%u, expected n_writes=%u!\n",
219 			 barrier_nr, set_size, expect_size);
220 		goto bail;
221 	}
222 
223 	/* Clean up list of requests processed during current epoch. */
224 	/* this extra list walk restart is paranoia,
225 	 * to catch requests being barrier-acked "unexpectedly".
226 	 * It usually should find the same req again, or some READ preceding it. */
227 	list_for_each_entry(req, &connection->transfer_log, tl_requests)
228 		if (req->epoch == expect_epoch) {
229 			tmp = req;
230 			break;
231 		}
232 	req = list_prepare_entry(tmp, &connection->transfer_log, tl_requests);
233 	list_for_each_entry_safe_from(req, r, &connection->transfer_log, tl_requests) {
234 		if (req->epoch != expect_epoch)
235 			break;
236 		_req_mod(req, BARRIER_ACKED);
237 	}
238 	spin_unlock_irq(&connection->resource->req_lock);
239 
240 	return;
241 
242 bail:
243 	spin_unlock_irq(&connection->resource->req_lock);
244 	conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
245 }
246 
247 
248 /**
249  * _tl_restart() - Walks the transfer log, and applies an action to all requests
250  * @connection:	DRBD connection to operate on.
251  * @what:       The action/event to perform with all request objects
252  *
253  * @what might be one of CONNECTION_LOST_WHILE_PENDING, RESEND, FAIL_FROZEN_DISK_IO,
254  * RESTART_FROZEN_DISK_IO.
255  */
256 /* must hold resource->req_lock */
_tl_restart(struct drbd_connection * connection,enum drbd_req_event what)257 void _tl_restart(struct drbd_connection *connection, enum drbd_req_event what)
258 {
259 	struct drbd_request *req, *r;
260 
261 	list_for_each_entry_safe(req, r, &connection->transfer_log, tl_requests)
262 		_req_mod(req, what);
263 }
264 
tl_restart(struct drbd_connection * connection,enum drbd_req_event what)265 void tl_restart(struct drbd_connection *connection, enum drbd_req_event what)
266 {
267 	spin_lock_irq(&connection->resource->req_lock);
268 	_tl_restart(connection, what);
269 	spin_unlock_irq(&connection->resource->req_lock);
270 }
271 
272 /**
273  * tl_clear() - Clears all requests and &struct drbd_tl_epoch objects out of the TL
274  * @connection:	DRBD connection.
275  *
276  * This is called after the connection to the peer was lost. The storage covered
277  * by the requests on the transfer gets marked as our of sync. Called from the
278  * receiver thread and the worker thread.
279  */
tl_clear(struct drbd_connection * connection)280 void tl_clear(struct drbd_connection *connection)
281 {
282 	tl_restart(connection, CONNECTION_LOST_WHILE_PENDING);
283 }
284 
285 /**
286  * tl_abort_disk_io() - Abort disk I/O for all requests for a certain device in the TL
287  * @device:	DRBD device.
288  */
tl_abort_disk_io(struct drbd_device * device)289 void tl_abort_disk_io(struct drbd_device *device)
290 {
291 	struct drbd_connection *connection = first_peer_device(device)->connection;
292 	struct drbd_request *req, *r;
293 
294 	spin_lock_irq(&connection->resource->req_lock);
295 	list_for_each_entry_safe(req, r, &connection->transfer_log, tl_requests) {
296 		if (!(req->rq_state & RQ_LOCAL_PENDING))
297 			continue;
298 		if (req->device != device)
299 			continue;
300 		_req_mod(req, ABORT_DISK_IO);
301 	}
302 	spin_unlock_irq(&connection->resource->req_lock);
303 }
304 
drbd_thread_setup(void * arg)305 static int drbd_thread_setup(void *arg)
306 {
307 	struct drbd_thread *thi = (struct drbd_thread *) arg;
308 	struct drbd_resource *resource = thi->resource;
309 	unsigned long flags;
310 	int retval;
311 
312 	snprintf(current->comm, sizeof(current->comm), "drbd_%c_%s",
313 		 thi->name[0],
314 		 resource->name);
315 
316 	allow_kernel_signal(DRBD_SIGKILL);
317 	allow_kernel_signal(SIGXCPU);
318 restart:
319 	retval = thi->function(thi);
320 
321 	spin_lock_irqsave(&thi->t_lock, flags);
322 
323 	/* if the receiver has been "EXITING", the last thing it did
324 	 * was set the conn state to "StandAlone",
325 	 * if now a re-connect request comes in, conn state goes C_UNCONNECTED,
326 	 * and receiver thread will be "started".
327 	 * drbd_thread_start needs to set "RESTARTING" in that case.
328 	 * t_state check and assignment needs to be within the same spinlock,
329 	 * so either thread_start sees EXITING, and can remap to RESTARTING,
330 	 * or thread_start see NONE, and can proceed as normal.
331 	 */
332 
333 	if (thi->t_state == RESTARTING) {
334 		drbd_info(resource, "Restarting %s thread\n", thi->name);
335 		thi->t_state = RUNNING;
336 		spin_unlock_irqrestore(&thi->t_lock, flags);
337 		goto restart;
338 	}
339 
340 	thi->task = NULL;
341 	thi->t_state = NONE;
342 	smp_mb();
343 	complete_all(&thi->stop);
344 	spin_unlock_irqrestore(&thi->t_lock, flags);
345 
346 	drbd_info(resource, "Terminating %s\n", current->comm);
347 
348 	/* Release mod reference taken when thread was started */
349 
350 	if (thi->connection)
351 		kref_put(&thi->connection->kref, drbd_destroy_connection);
352 	kref_put(&resource->kref, drbd_destroy_resource);
353 	module_put(THIS_MODULE);
354 	return retval;
355 }
356 
drbd_thread_init(struct drbd_resource * resource,struct drbd_thread * thi,int (* func)(struct drbd_thread *),const char * name)357 static void drbd_thread_init(struct drbd_resource *resource, struct drbd_thread *thi,
358 			     int (*func) (struct drbd_thread *), const char *name)
359 {
360 	spin_lock_init(&thi->t_lock);
361 	thi->task    = NULL;
362 	thi->t_state = NONE;
363 	thi->function = func;
364 	thi->resource = resource;
365 	thi->connection = NULL;
366 	thi->name = name;
367 }
368 
drbd_thread_start(struct drbd_thread * thi)369 int drbd_thread_start(struct drbd_thread *thi)
370 {
371 	struct drbd_resource *resource = thi->resource;
372 	struct task_struct *nt;
373 	unsigned long flags;
374 
375 	/* is used from state engine doing drbd_thread_stop_nowait,
376 	 * while holding the req lock irqsave */
377 	spin_lock_irqsave(&thi->t_lock, flags);
378 
379 	switch (thi->t_state) {
380 	case NONE:
381 		drbd_info(resource, "Starting %s thread (from %s [%d])\n",
382 			 thi->name, current->comm, current->pid);
383 
384 		/* Get ref on module for thread - this is released when thread exits */
385 		if (!try_module_get(THIS_MODULE)) {
386 			drbd_err(resource, "Failed to get module reference in drbd_thread_start\n");
387 			spin_unlock_irqrestore(&thi->t_lock, flags);
388 			return false;
389 		}
390 
391 		kref_get(&resource->kref);
392 		if (thi->connection)
393 			kref_get(&thi->connection->kref);
394 
395 		init_completion(&thi->stop);
396 		thi->reset_cpu_mask = 1;
397 		thi->t_state = RUNNING;
398 		spin_unlock_irqrestore(&thi->t_lock, flags);
399 		flush_signals(current); /* otherw. may get -ERESTARTNOINTR */
400 
401 		nt = kthread_create(drbd_thread_setup, (void *) thi,
402 				    "drbd_%c_%s", thi->name[0], thi->resource->name);
403 
404 		if (IS_ERR(nt)) {
405 			drbd_err(resource, "Couldn't start thread\n");
406 
407 			if (thi->connection)
408 				kref_put(&thi->connection->kref, drbd_destroy_connection);
409 			kref_put(&resource->kref, drbd_destroy_resource);
410 			module_put(THIS_MODULE);
411 			return false;
412 		}
413 		spin_lock_irqsave(&thi->t_lock, flags);
414 		thi->task = nt;
415 		thi->t_state = RUNNING;
416 		spin_unlock_irqrestore(&thi->t_lock, flags);
417 		wake_up_process(nt);
418 		break;
419 	case EXITING:
420 		thi->t_state = RESTARTING;
421 		drbd_info(resource, "Restarting %s thread (from %s [%d])\n",
422 				thi->name, current->comm, current->pid);
423 		fallthrough;
424 	case RUNNING:
425 	case RESTARTING:
426 	default:
427 		spin_unlock_irqrestore(&thi->t_lock, flags);
428 		break;
429 	}
430 
431 	return true;
432 }
433 
434 
_drbd_thread_stop(struct drbd_thread * thi,int restart,int wait)435 void _drbd_thread_stop(struct drbd_thread *thi, int restart, int wait)
436 {
437 	unsigned long flags;
438 
439 	enum drbd_thread_state ns = restart ? RESTARTING : EXITING;
440 
441 	/* may be called from state engine, holding the req lock irqsave */
442 	spin_lock_irqsave(&thi->t_lock, flags);
443 
444 	if (thi->t_state == NONE) {
445 		spin_unlock_irqrestore(&thi->t_lock, flags);
446 		if (restart)
447 			drbd_thread_start(thi);
448 		return;
449 	}
450 
451 	if (thi->t_state != ns) {
452 		if (thi->task == NULL) {
453 			spin_unlock_irqrestore(&thi->t_lock, flags);
454 			return;
455 		}
456 
457 		thi->t_state = ns;
458 		smp_mb();
459 		init_completion(&thi->stop);
460 		if (thi->task != current)
461 			send_sig(DRBD_SIGKILL, thi->task, 1);
462 	}
463 
464 	spin_unlock_irqrestore(&thi->t_lock, flags);
465 
466 	if (wait)
467 		wait_for_completion(&thi->stop);
468 }
469 
conn_lowest_minor(struct drbd_connection * connection)470 int conn_lowest_minor(struct drbd_connection *connection)
471 {
472 	struct drbd_peer_device *peer_device;
473 	int vnr = 0, minor = -1;
474 
475 	rcu_read_lock();
476 	peer_device = idr_get_next(&connection->peer_devices, &vnr);
477 	if (peer_device)
478 		minor = device_to_minor(peer_device->device);
479 	rcu_read_unlock();
480 
481 	return minor;
482 }
483 
484 #ifdef CONFIG_SMP
485 /*
486  * drbd_calc_cpu_mask() - Generate CPU masks, spread over all CPUs
487  *
488  * Forces all threads of a resource onto the same CPU. This is beneficial for
489  * DRBD's performance. May be overwritten by user's configuration.
490  */
drbd_calc_cpu_mask(cpumask_var_t * cpu_mask)491 static void drbd_calc_cpu_mask(cpumask_var_t *cpu_mask)
492 {
493 	unsigned int *resources_per_cpu, min_index = ~0;
494 
495 	resources_per_cpu = kcalloc(nr_cpu_ids, sizeof(*resources_per_cpu),
496 				    GFP_KERNEL);
497 	if (resources_per_cpu) {
498 		struct drbd_resource *resource;
499 		unsigned int cpu, min = ~0;
500 
501 		rcu_read_lock();
502 		for_each_resource_rcu(resource, &drbd_resources) {
503 			for_each_cpu(cpu, resource->cpu_mask)
504 				resources_per_cpu[cpu]++;
505 		}
506 		rcu_read_unlock();
507 		for_each_online_cpu(cpu) {
508 			if (resources_per_cpu[cpu] < min) {
509 				min = resources_per_cpu[cpu];
510 				min_index = cpu;
511 			}
512 		}
513 		kfree(resources_per_cpu);
514 	}
515 	if (min_index == ~0) {
516 		cpumask_setall(*cpu_mask);
517 		return;
518 	}
519 	cpumask_set_cpu(min_index, *cpu_mask);
520 }
521 
522 /**
523  * drbd_thread_current_set_cpu() - modifies the cpu mask of the _current_ thread
524  * @thi:	drbd_thread object
525  *
526  * call in the "main loop" of _all_ threads, no need for any mutex, current won't die
527  * prematurely.
528  */
drbd_thread_current_set_cpu(struct drbd_thread * thi)529 void drbd_thread_current_set_cpu(struct drbd_thread *thi)
530 {
531 	struct drbd_resource *resource = thi->resource;
532 	struct task_struct *p = current;
533 
534 	if (!thi->reset_cpu_mask)
535 		return;
536 	thi->reset_cpu_mask = 0;
537 	set_cpus_allowed_ptr(p, resource->cpu_mask);
538 }
539 #else
540 #define drbd_calc_cpu_mask(A) ({})
541 #endif
542 
543 /*
544  * drbd_header_size  -  size of a packet header
545  *
546  * The header size is a multiple of 8, so any payload following the header is
547  * word aligned on 64-bit architectures.  (The bitmap send and receive code
548  * relies on this.)
549  */
drbd_header_size(struct drbd_connection * connection)550 unsigned int drbd_header_size(struct drbd_connection *connection)
551 {
552 	if (connection->agreed_pro_version >= 100) {
553 		BUILD_BUG_ON(!IS_ALIGNED(sizeof(struct p_header100), 8));
554 		return sizeof(struct p_header100);
555 	} else {
556 		BUILD_BUG_ON(sizeof(struct p_header80) !=
557 			     sizeof(struct p_header95));
558 		BUILD_BUG_ON(!IS_ALIGNED(sizeof(struct p_header80), 8));
559 		return sizeof(struct p_header80);
560 	}
561 }
562 
prepare_header80(struct p_header80 * h,enum drbd_packet cmd,int size)563 static unsigned int prepare_header80(struct p_header80 *h, enum drbd_packet cmd, int size)
564 {
565 	h->magic   = cpu_to_be32(DRBD_MAGIC);
566 	h->command = cpu_to_be16(cmd);
567 	h->length  = cpu_to_be16(size);
568 	return sizeof(struct p_header80);
569 }
570 
prepare_header95(struct p_header95 * h,enum drbd_packet cmd,int size)571 static unsigned int prepare_header95(struct p_header95 *h, enum drbd_packet cmd, int size)
572 {
573 	h->magic   = cpu_to_be16(DRBD_MAGIC_BIG);
574 	h->command = cpu_to_be16(cmd);
575 	h->length = cpu_to_be32(size);
576 	return sizeof(struct p_header95);
577 }
578 
prepare_header100(struct p_header100 * h,enum drbd_packet cmd,int size,int vnr)579 static unsigned int prepare_header100(struct p_header100 *h, enum drbd_packet cmd,
580 				      int size, int vnr)
581 {
582 	h->magic = cpu_to_be32(DRBD_MAGIC_100);
583 	h->volume = cpu_to_be16(vnr);
584 	h->command = cpu_to_be16(cmd);
585 	h->length = cpu_to_be32(size);
586 	h->pad = 0;
587 	return sizeof(struct p_header100);
588 }
589 
prepare_header(struct drbd_connection * connection,int vnr,void * buffer,enum drbd_packet cmd,int size)590 static unsigned int prepare_header(struct drbd_connection *connection, int vnr,
591 				   void *buffer, enum drbd_packet cmd, int size)
592 {
593 	if (connection->agreed_pro_version >= 100)
594 		return prepare_header100(buffer, cmd, size, vnr);
595 	else if (connection->agreed_pro_version >= 95 &&
596 		 size > DRBD_MAX_SIZE_H80_PACKET)
597 		return prepare_header95(buffer, cmd, size);
598 	else
599 		return prepare_header80(buffer, cmd, size);
600 }
601 
__conn_prepare_command(struct drbd_connection * connection,struct drbd_socket * sock)602 static void *__conn_prepare_command(struct drbd_connection *connection,
603 				    struct drbd_socket *sock)
604 {
605 	if (!sock->socket)
606 		return NULL;
607 	return sock->sbuf + drbd_header_size(connection);
608 }
609 
conn_prepare_command(struct drbd_connection * connection,struct drbd_socket * sock)610 void *conn_prepare_command(struct drbd_connection *connection, struct drbd_socket *sock)
611 {
612 	void *p;
613 
614 	mutex_lock(&sock->mutex);
615 	p = __conn_prepare_command(connection, sock);
616 	if (!p)
617 		mutex_unlock(&sock->mutex);
618 
619 	return p;
620 }
621 
drbd_prepare_command(struct drbd_peer_device * peer_device,struct drbd_socket * sock)622 void *drbd_prepare_command(struct drbd_peer_device *peer_device, struct drbd_socket *sock)
623 {
624 	return conn_prepare_command(peer_device->connection, sock);
625 }
626 
__send_command(struct drbd_connection * connection,int vnr,struct drbd_socket * sock,enum drbd_packet cmd,unsigned int header_size,void * data,unsigned int size)627 static int __send_command(struct drbd_connection *connection, int vnr,
628 			  struct drbd_socket *sock, enum drbd_packet cmd,
629 			  unsigned int header_size, void *data,
630 			  unsigned int size)
631 {
632 	int msg_flags;
633 	int err;
634 
635 	/*
636 	 * Called with @data == NULL and the size of the data blocks in @size
637 	 * for commands that send data blocks.  For those commands, omit the
638 	 * MSG_MORE flag: this will increase the likelihood that data blocks
639 	 * which are page aligned on the sender will end up page aligned on the
640 	 * receiver.
641 	 */
642 	msg_flags = data ? MSG_MORE : 0;
643 
644 	header_size += prepare_header(connection, vnr, sock->sbuf, cmd,
645 				      header_size + size);
646 	err = drbd_send_all(connection, sock->socket, sock->sbuf, header_size,
647 			    msg_flags);
648 	if (data && !err)
649 		err = drbd_send_all(connection, sock->socket, data, size, 0);
650 	/* DRBD protocol "pings" are latency critical.
651 	 * This is supposed to trigger tcp_push_pending_frames() */
652 	if (!err && (cmd == P_PING || cmd == P_PING_ACK))
653 		tcp_sock_set_nodelay(sock->socket->sk);
654 
655 	return err;
656 }
657 
__conn_send_command(struct drbd_connection * connection,struct drbd_socket * sock,enum drbd_packet cmd,unsigned int header_size,void * data,unsigned int size)658 static int __conn_send_command(struct drbd_connection *connection, struct drbd_socket *sock,
659 			       enum drbd_packet cmd, unsigned int header_size,
660 			       void *data, unsigned int size)
661 {
662 	return __send_command(connection, 0, sock, cmd, header_size, data, size);
663 }
664 
conn_send_command(struct drbd_connection * connection,struct drbd_socket * sock,enum drbd_packet cmd,unsigned int header_size,void * data,unsigned int size)665 int conn_send_command(struct drbd_connection *connection, struct drbd_socket *sock,
666 		      enum drbd_packet cmd, unsigned int header_size,
667 		      void *data, unsigned int size)
668 {
669 	int err;
670 
671 	err = __conn_send_command(connection, sock, cmd, header_size, data, size);
672 	mutex_unlock(&sock->mutex);
673 	return err;
674 }
675 
drbd_send_command(struct drbd_peer_device * peer_device,struct drbd_socket * sock,enum drbd_packet cmd,unsigned int header_size,void * data,unsigned int size)676 int drbd_send_command(struct drbd_peer_device *peer_device, struct drbd_socket *sock,
677 		      enum drbd_packet cmd, unsigned int header_size,
678 		      void *data, unsigned int size)
679 {
680 	int err;
681 
682 	err = __send_command(peer_device->connection, peer_device->device->vnr,
683 			     sock, cmd, header_size, data, size);
684 	mutex_unlock(&sock->mutex);
685 	return err;
686 }
687 
drbd_send_ping(struct drbd_connection * connection)688 int drbd_send_ping(struct drbd_connection *connection)
689 {
690 	struct drbd_socket *sock;
691 
692 	sock = &connection->meta;
693 	if (!conn_prepare_command(connection, sock))
694 		return -EIO;
695 	return conn_send_command(connection, sock, P_PING, 0, NULL, 0);
696 }
697 
drbd_send_ping_ack(struct drbd_connection * connection)698 int drbd_send_ping_ack(struct drbd_connection *connection)
699 {
700 	struct drbd_socket *sock;
701 
702 	sock = &connection->meta;
703 	if (!conn_prepare_command(connection, sock))
704 		return -EIO;
705 	return conn_send_command(connection, sock, P_PING_ACK, 0, NULL, 0);
706 }
707 
drbd_send_sync_param(struct drbd_peer_device * peer_device)708 int drbd_send_sync_param(struct drbd_peer_device *peer_device)
709 {
710 	struct drbd_socket *sock;
711 	struct p_rs_param_95 *p;
712 	int size;
713 	const int apv = peer_device->connection->agreed_pro_version;
714 	enum drbd_packet cmd;
715 	struct net_conf *nc;
716 	struct disk_conf *dc;
717 
718 	sock = &peer_device->connection->data;
719 	p = drbd_prepare_command(peer_device, sock);
720 	if (!p)
721 		return -EIO;
722 
723 	rcu_read_lock();
724 	nc = rcu_dereference(peer_device->connection->net_conf);
725 
726 	size = apv <= 87 ? sizeof(struct p_rs_param)
727 		: apv == 88 ? sizeof(struct p_rs_param)
728 			+ strlen(nc->verify_alg) + 1
729 		: apv <= 94 ? sizeof(struct p_rs_param_89)
730 		: /* apv >= 95 */ sizeof(struct p_rs_param_95);
731 
732 	cmd = apv >= 89 ? P_SYNC_PARAM89 : P_SYNC_PARAM;
733 
734 	/* initialize verify_alg and csums_alg */
735 	memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
736 
737 	if (get_ldev(peer_device->device)) {
738 		dc = rcu_dereference(peer_device->device->ldev->disk_conf);
739 		p->resync_rate = cpu_to_be32(dc->resync_rate);
740 		p->c_plan_ahead = cpu_to_be32(dc->c_plan_ahead);
741 		p->c_delay_target = cpu_to_be32(dc->c_delay_target);
742 		p->c_fill_target = cpu_to_be32(dc->c_fill_target);
743 		p->c_max_rate = cpu_to_be32(dc->c_max_rate);
744 		put_ldev(peer_device->device);
745 	} else {
746 		p->resync_rate = cpu_to_be32(DRBD_RESYNC_RATE_DEF);
747 		p->c_plan_ahead = cpu_to_be32(DRBD_C_PLAN_AHEAD_DEF);
748 		p->c_delay_target = cpu_to_be32(DRBD_C_DELAY_TARGET_DEF);
749 		p->c_fill_target = cpu_to_be32(DRBD_C_FILL_TARGET_DEF);
750 		p->c_max_rate = cpu_to_be32(DRBD_C_MAX_RATE_DEF);
751 	}
752 
753 	if (apv >= 88)
754 		strcpy(p->verify_alg, nc->verify_alg);
755 	if (apv >= 89)
756 		strcpy(p->csums_alg, nc->csums_alg);
757 	rcu_read_unlock();
758 
759 	return drbd_send_command(peer_device, sock, cmd, size, NULL, 0);
760 }
761 
__drbd_send_protocol(struct drbd_connection * connection,enum drbd_packet cmd)762 int __drbd_send_protocol(struct drbd_connection *connection, enum drbd_packet cmd)
763 {
764 	struct drbd_socket *sock;
765 	struct p_protocol *p;
766 	struct net_conf *nc;
767 	int size, cf;
768 
769 	sock = &connection->data;
770 	p = __conn_prepare_command(connection, sock);
771 	if (!p)
772 		return -EIO;
773 
774 	rcu_read_lock();
775 	nc = rcu_dereference(connection->net_conf);
776 
777 	if (nc->tentative && connection->agreed_pro_version < 92) {
778 		rcu_read_unlock();
779 		drbd_err(connection, "--dry-run is not supported by peer");
780 		return -EOPNOTSUPP;
781 	}
782 
783 	size = sizeof(*p);
784 	if (connection->agreed_pro_version >= 87)
785 		size += strlen(nc->integrity_alg) + 1;
786 
787 	p->protocol      = cpu_to_be32(nc->wire_protocol);
788 	p->after_sb_0p   = cpu_to_be32(nc->after_sb_0p);
789 	p->after_sb_1p   = cpu_to_be32(nc->after_sb_1p);
790 	p->after_sb_2p   = cpu_to_be32(nc->after_sb_2p);
791 	p->two_primaries = cpu_to_be32(nc->two_primaries);
792 	cf = 0;
793 	if (nc->discard_my_data)
794 		cf |= CF_DISCARD_MY_DATA;
795 	if (nc->tentative)
796 		cf |= CF_DRY_RUN;
797 	p->conn_flags    = cpu_to_be32(cf);
798 
799 	if (connection->agreed_pro_version >= 87)
800 		strcpy(p->integrity_alg, nc->integrity_alg);
801 	rcu_read_unlock();
802 
803 	return __conn_send_command(connection, sock, cmd, size, NULL, 0);
804 }
805 
drbd_send_protocol(struct drbd_connection * connection)806 int drbd_send_protocol(struct drbd_connection *connection)
807 {
808 	int err;
809 
810 	mutex_lock(&connection->data.mutex);
811 	err = __drbd_send_protocol(connection, P_PROTOCOL);
812 	mutex_unlock(&connection->data.mutex);
813 
814 	return err;
815 }
816 
_drbd_send_uuids(struct drbd_peer_device * peer_device,u64 uuid_flags)817 static int _drbd_send_uuids(struct drbd_peer_device *peer_device, u64 uuid_flags)
818 {
819 	struct drbd_device *device = peer_device->device;
820 	struct drbd_socket *sock;
821 	struct p_uuids *p;
822 	int i;
823 
824 	if (!get_ldev_if_state(device, D_NEGOTIATING))
825 		return 0;
826 
827 	sock = &peer_device->connection->data;
828 	p = drbd_prepare_command(peer_device, sock);
829 	if (!p) {
830 		put_ldev(device);
831 		return -EIO;
832 	}
833 	spin_lock_irq(&device->ldev->md.uuid_lock);
834 	for (i = UI_CURRENT; i < UI_SIZE; i++)
835 		p->uuid[i] = cpu_to_be64(device->ldev->md.uuid[i]);
836 	spin_unlock_irq(&device->ldev->md.uuid_lock);
837 
838 	device->comm_bm_set = drbd_bm_total_weight(device);
839 	p->uuid[UI_SIZE] = cpu_to_be64(device->comm_bm_set);
840 	rcu_read_lock();
841 	uuid_flags |= rcu_dereference(peer_device->connection->net_conf)->discard_my_data ? 1 : 0;
842 	rcu_read_unlock();
843 	uuid_flags |= test_bit(CRASHED_PRIMARY, &device->flags) ? 2 : 0;
844 	uuid_flags |= device->new_state_tmp.disk == D_INCONSISTENT ? 4 : 0;
845 	p->uuid[UI_FLAGS] = cpu_to_be64(uuid_flags);
846 
847 	put_ldev(device);
848 	return drbd_send_command(peer_device, sock, P_UUIDS, sizeof(*p), NULL, 0);
849 }
850 
drbd_send_uuids(struct drbd_peer_device * peer_device)851 int drbd_send_uuids(struct drbd_peer_device *peer_device)
852 {
853 	return _drbd_send_uuids(peer_device, 0);
854 }
855 
drbd_send_uuids_skip_initial_sync(struct drbd_peer_device * peer_device)856 int drbd_send_uuids_skip_initial_sync(struct drbd_peer_device *peer_device)
857 {
858 	return _drbd_send_uuids(peer_device, 8);
859 }
860 
drbd_print_uuids(struct drbd_device * device,const char * text)861 void drbd_print_uuids(struct drbd_device *device, const char *text)
862 {
863 	if (get_ldev_if_state(device, D_NEGOTIATING)) {
864 		u64 *uuid = device->ldev->md.uuid;
865 		drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX\n",
866 		     text,
867 		     (unsigned long long)uuid[UI_CURRENT],
868 		     (unsigned long long)uuid[UI_BITMAP],
869 		     (unsigned long long)uuid[UI_HISTORY_START],
870 		     (unsigned long long)uuid[UI_HISTORY_END]);
871 		put_ldev(device);
872 	} else {
873 		drbd_info(device, "%s effective data uuid: %016llX\n",
874 				text,
875 				(unsigned long long)device->ed_uuid);
876 	}
877 }
878 
drbd_gen_and_send_sync_uuid(struct drbd_peer_device * peer_device)879 void drbd_gen_and_send_sync_uuid(struct drbd_peer_device *peer_device)
880 {
881 	struct drbd_device *device = peer_device->device;
882 	struct drbd_socket *sock;
883 	struct p_rs_uuid *p;
884 	u64 uuid;
885 
886 	D_ASSERT(device, device->state.disk == D_UP_TO_DATE);
887 
888 	uuid = device->ldev->md.uuid[UI_BITMAP];
889 	if (uuid && uuid != UUID_JUST_CREATED)
890 		uuid = uuid + UUID_NEW_BM_OFFSET;
891 	else
892 		get_random_bytes(&uuid, sizeof(u64));
893 	drbd_uuid_set(device, UI_BITMAP, uuid);
894 	drbd_print_uuids(device, "updated sync UUID");
895 	drbd_md_sync(device);
896 
897 	sock = &peer_device->connection->data;
898 	p = drbd_prepare_command(peer_device, sock);
899 	if (p) {
900 		p->uuid = cpu_to_be64(uuid);
901 		drbd_send_command(peer_device, sock, P_SYNC_UUID, sizeof(*p), NULL, 0);
902 	}
903 }
904 
905 /* communicated if (agreed_features & DRBD_FF_WSAME) */
906 static void
assign_p_sizes_qlim(struct drbd_device * device,struct p_sizes * p,struct request_queue * q)907 assign_p_sizes_qlim(struct drbd_device *device, struct p_sizes *p,
908 					struct request_queue *q)
909 {
910 	if (q) {
911 		p->qlim->physical_block_size = cpu_to_be32(queue_physical_block_size(q));
912 		p->qlim->logical_block_size = cpu_to_be32(queue_logical_block_size(q));
913 		p->qlim->alignment_offset = cpu_to_be32(queue_alignment_offset(q));
914 		p->qlim->io_min = cpu_to_be32(queue_io_min(q));
915 		p->qlim->io_opt = cpu_to_be32(queue_io_opt(q));
916 		p->qlim->discard_enabled = blk_queue_discard(q);
917 		p->qlim->write_same_capable = !!q->limits.max_write_same_sectors;
918 	} else {
919 		q = device->rq_queue;
920 		p->qlim->physical_block_size = cpu_to_be32(queue_physical_block_size(q));
921 		p->qlim->logical_block_size = cpu_to_be32(queue_logical_block_size(q));
922 		p->qlim->alignment_offset = 0;
923 		p->qlim->io_min = cpu_to_be32(queue_io_min(q));
924 		p->qlim->io_opt = cpu_to_be32(queue_io_opt(q));
925 		p->qlim->discard_enabled = 0;
926 		p->qlim->write_same_capable = 0;
927 	}
928 }
929 
drbd_send_sizes(struct drbd_peer_device * peer_device,int trigger_reply,enum dds_flags flags)930 int drbd_send_sizes(struct drbd_peer_device *peer_device, int trigger_reply, enum dds_flags flags)
931 {
932 	struct drbd_device *device = peer_device->device;
933 	struct drbd_socket *sock;
934 	struct p_sizes *p;
935 	sector_t d_size, u_size;
936 	int q_order_type;
937 	unsigned int max_bio_size;
938 	unsigned int packet_size;
939 
940 	sock = &peer_device->connection->data;
941 	p = drbd_prepare_command(peer_device, sock);
942 	if (!p)
943 		return -EIO;
944 
945 	packet_size = sizeof(*p);
946 	if (peer_device->connection->agreed_features & DRBD_FF_WSAME)
947 		packet_size += sizeof(p->qlim[0]);
948 
949 	memset(p, 0, packet_size);
950 	if (get_ldev_if_state(device, D_NEGOTIATING)) {
951 		struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
952 		d_size = drbd_get_max_capacity(device->ldev);
953 		rcu_read_lock();
954 		u_size = rcu_dereference(device->ldev->disk_conf)->disk_size;
955 		rcu_read_unlock();
956 		q_order_type = drbd_queue_order_type(device);
957 		max_bio_size = queue_max_hw_sectors(q) << 9;
958 		max_bio_size = min(max_bio_size, DRBD_MAX_BIO_SIZE);
959 		assign_p_sizes_qlim(device, p, q);
960 		put_ldev(device);
961 	} else {
962 		d_size = 0;
963 		u_size = 0;
964 		q_order_type = QUEUE_ORDERED_NONE;
965 		max_bio_size = DRBD_MAX_BIO_SIZE; /* ... multiple BIOs per peer_request */
966 		assign_p_sizes_qlim(device, p, NULL);
967 	}
968 
969 	if (peer_device->connection->agreed_pro_version <= 94)
970 		max_bio_size = min(max_bio_size, DRBD_MAX_SIZE_H80_PACKET);
971 	else if (peer_device->connection->agreed_pro_version < 100)
972 		max_bio_size = min(max_bio_size, DRBD_MAX_BIO_SIZE_P95);
973 
974 	p->d_size = cpu_to_be64(d_size);
975 	p->u_size = cpu_to_be64(u_size);
976 	if (trigger_reply)
977 		p->c_size = 0;
978 	else
979 		p->c_size = cpu_to_be64(get_capacity(device->vdisk));
980 	p->max_bio_size = cpu_to_be32(max_bio_size);
981 	p->queue_order_type = cpu_to_be16(q_order_type);
982 	p->dds_flags = cpu_to_be16(flags);
983 
984 	return drbd_send_command(peer_device, sock, P_SIZES, packet_size, NULL, 0);
985 }
986 
987 /**
988  * drbd_send_current_state() - Sends the drbd state to the peer
989  * @peer_device:	DRBD peer device.
990  */
drbd_send_current_state(struct drbd_peer_device * peer_device)991 int drbd_send_current_state(struct drbd_peer_device *peer_device)
992 {
993 	struct drbd_socket *sock;
994 	struct p_state *p;
995 
996 	sock = &peer_device->connection->data;
997 	p = drbd_prepare_command(peer_device, sock);
998 	if (!p)
999 		return -EIO;
1000 	p->state = cpu_to_be32(peer_device->device->state.i); /* Within the send mutex */
1001 	return drbd_send_command(peer_device, sock, P_STATE, sizeof(*p), NULL, 0);
1002 }
1003 
1004 /**
1005  * drbd_send_state() - After a state change, sends the new state to the peer
1006  * @peer_device:      DRBD peer device.
1007  * @state:     the state to send, not necessarily the current state.
1008  *
1009  * Each state change queues an "after_state_ch" work, which will eventually
1010  * send the resulting new state to the peer. If more state changes happen
1011  * between queuing and processing of the after_state_ch work, we still
1012  * want to send each intermediary state in the order it occurred.
1013  */
drbd_send_state(struct drbd_peer_device * peer_device,union drbd_state state)1014 int drbd_send_state(struct drbd_peer_device *peer_device, union drbd_state state)
1015 {
1016 	struct drbd_socket *sock;
1017 	struct p_state *p;
1018 
1019 	sock = &peer_device->connection->data;
1020 	p = drbd_prepare_command(peer_device, sock);
1021 	if (!p)
1022 		return -EIO;
1023 	p->state = cpu_to_be32(state.i); /* Within the send mutex */
1024 	return drbd_send_command(peer_device, sock, P_STATE, sizeof(*p), NULL, 0);
1025 }
1026 
drbd_send_state_req(struct drbd_peer_device * peer_device,union drbd_state mask,union drbd_state val)1027 int drbd_send_state_req(struct drbd_peer_device *peer_device, union drbd_state mask, union drbd_state val)
1028 {
1029 	struct drbd_socket *sock;
1030 	struct p_req_state *p;
1031 
1032 	sock = &peer_device->connection->data;
1033 	p = drbd_prepare_command(peer_device, sock);
1034 	if (!p)
1035 		return -EIO;
1036 	p->mask = cpu_to_be32(mask.i);
1037 	p->val = cpu_to_be32(val.i);
1038 	return drbd_send_command(peer_device, sock, P_STATE_CHG_REQ, sizeof(*p), NULL, 0);
1039 }
1040 
conn_send_state_req(struct drbd_connection * connection,union drbd_state mask,union drbd_state val)1041 int conn_send_state_req(struct drbd_connection *connection, union drbd_state mask, union drbd_state val)
1042 {
1043 	enum drbd_packet cmd;
1044 	struct drbd_socket *sock;
1045 	struct p_req_state *p;
1046 
1047 	cmd = connection->agreed_pro_version < 100 ? P_STATE_CHG_REQ : P_CONN_ST_CHG_REQ;
1048 	sock = &connection->data;
1049 	p = conn_prepare_command(connection, sock);
1050 	if (!p)
1051 		return -EIO;
1052 	p->mask = cpu_to_be32(mask.i);
1053 	p->val = cpu_to_be32(val.i);
1054 	return conn_send_command(connection, sock, cmd, sizeof(*p), NULL, 0);
1055 }
1056 
drbd_send_sr_reply(struct drbd_peer_device * peer_device,enum drbd_state_rv retcode)1057 void drbd_send_sr_reply(struct drbd_peer_device *peer_device, enum drbd_state_rv retcode)
1058 {
1059 	struct drbd_socket *sock;
1060 	struct p_req_state_reply *p;
1061 
1062 	sock = &peer_device->connection->meta;
1063 	p = drbd_prepare_command(peer_device, sock);
1064 	if (p) {
1065 		p->retcode = cpu_to_be32(retcode);
1066 		drbd_send_command(peer_device, sock, P_STATE_CHG_REPLY, sizeof(*p), NULL, 0);
1067 	}
1068 }
1069 
conn_send_sr_reply(struct drbd_connection * connection,enum drbd_state_rv retcode)1070 void conn_send_sr_reply(struct drbd_connection *connection, enum drbd_state_rv retcode)
1071 {
1072 	struct drbd_socket *sock;
1073 	struct p_req_state_reply *p;
1074 	enum drbd_packet cmd = connection->agreed_pro_version < 100 ? P_STATE_CHG_REPLY : P_CONN_ST_CHG_REPLY;
1075 
1076 	sock = &connection->meta;
1077 	p = conn_prepare_command(connection, sock);
1078 	if (p) {
1079 		p->retcode = cpu_to_be32(retcode);
1080 		conn_send_command(connection, sock, cmd, sizeof(*p), NULL, 0);
1081 	}
1082 }
1083 
dcbp_set_code(struct p_compressed_bm * p,enum drbd_bitmap_code code)1084 static void dcbp_set_code(struct p_compressed_bm *p, enum drbd_bitmap_code code)
1085 {
1086 	BUG_ON(code & ~0xf);
1087 	p->encoding = (p->encoding & ~0xf) | code;
1088 }
1089 
dcbp_set_start(struct p_compressed_bm * p,int set)1090 static void dcbp_set_start(struct p_compressed_bm *p, int set)
1091 {
1092 	p->encoding = (p->encoding & ~0x80) | (set ? 0x80 : 0);
1093 }
1094 
dcbp_set_pad_bits(struct p_compressed_bm * p,int n)1095 static void dcbp_set_pad_bits(struct p_compressed_bm *p, int n)
1096 {
1097 	BUG_ON(n & ~0x7);
1098 	p->encoding = (p->encoding & (~0x7 << 4)) | (n << 4);
1099 }
1100 
fill_bitmap_rle_bits(struct drbd_device * device,struct p_compressed_bm * p,unsigned int size,struct bm_xfer_ctx * c)1101 static int fill_bitmap_rle_bits(struct drbd_device *device,
1102 			 struct p_compressed_bm *p,
1103 			 unsigned int size,
1104 			 struct bm_xfer_ctx *c)
1105 {
1106 	struct bitstream bs;
1107 	unsigned long plain_bits;
1108 	unsigned long tmp;
1109 	unsigned long rl;
1110 	unsigned len;
1111 	unsigned toggle;
1112 	int bits, use_rle;
1113 
1114 	/* may we use this feature? */
1115 	rcu_read_lock();
1116 	use_rle = rcu_dereference(first_peer_device(device)->connection->net_conf)->use_rle;
1117 	rcu_read_unlock();
1118 	if (!use_rle || first_peer_device(device)->connection->agreed_pro_version < 90)
1119 		return 0;
1120 
1121 	if (c->bit_offset >= c->bm_bits)
1122 		return 0; /* nothing to do. */
1123 
1124 	/* use at most thus many bytes */
1125 	bitstream_init(&bs, p->code, size, 0);
1126 	memset(p->code, 0, size);
1127 	/* plain bits covered in this code string */
1128 	plain_bits = 0;
1129 
1130 	/* p->encoding & 0x80 stores whether the first run length is set.
1131 	 * bit offset is implicit.
1132 	 * start with toggle == 2 to be able to tell the first iteration */
1133 	toggle = 2;
1134 
1135 	/* see how much plain bits we can stuff into one packet
1136 	 * using RLE and VLI. */
1137 	do {
1138 		tmp = (toggle == 0) ? _drbd_bm_find_next_zero(device, c->bit_offset)
1139 				    : _drbd_bm_find_next(device, c->bit_offset);
1140 		if (tmp == -1UL)
1141 			tmp = c->bm_bits;
1142 		rl = tmp - c->bit_offset;
1143 
1144 		if (toggle == 2) { /* first iteration */
1145 			if (rl == 0) {
1146 				/* the first checked bit was set,
1147 				 * store start value, */
1148 				dcbp_set_start(p, 1);
1149 				/* but skip encoding of zero run length */
1150 				toggle = !toggle;
1151 				continue;
1152 			}
1153 			dcbp_set_start(p, 0);
1154 		}
1155 
1156 		/* paranoia: catch zero runlength.
1157 		 * can only happen if bitmap is modified while we scan it. */
1158 		if (rl == 0) {
1159 			drbd_err(device, "unexpected zero runlength while encoding bitmap "
1160 			    "t:%u bo:%lu\n", toggle, c->bit_offset);
1161 			return -1;
1162 		}
1163 
1164 		bits = vli_encode_bits(&bs, rl);
1165 		if (bits == -ENOBUFS) /* buffer full */
1166 			break;
1167 		if (bits <= 0) {
1168 			drbd_err(device, "error while encoding bitmap: %d\n", bits);
1169 			return 0;
1170 		}
1171 
1172 		toggle = !toggle;
1173 		plain_bits += rl;
1174 		c->bit_offset = tmp;
1175 	} while (c->bit_offset < c->bm_bits);
1176 
1177 	len = bs.cur.b - p->code + !!bs.cur.bit;
1178 
1179 	if (plain_bits < (len << 3)) {
1180 		/* incompressible with this method.
1181 		 * we need to rewind both word and bit position. */
1182 		c->bit_offset -= plain_bits;
1183 		bm_xfer_ctx_bit_to_word_offset(c);
1184 		c->bit_offset = c->word_offset * BITS_PER_LONG;
1185 		return 0;
1186 	}
1187 
1188 	/* RLE + VLI was able to compress it just fine.
1189 	 * update c->word_offset. */
1190 	bm_xfer_ctx_bit_to_word_offset(c);
1191 
1192 	/* store pad_bits */
1193 	dcbp_set_pad_bits(p, (8 - bs.cur.bit) & 0x7);
1194 
1195 	return len;
1196 }
1197 
1198 /*
1199  * send_bitmap_rle_or_plain
1200  *
1201  * Return 0 when done, 1 when another iteration is needed, and a negative error
1202  * code upon failure.
1203  */
1204 static int
send_bitmap_rle_or_plain(struct drbd_device * device,struct bm_xfer_ctx * c)1205 send_bitmap_rle_or_plain(struct drbd_device *device, struct bm_xfer_ctx *c)
1206 {
1207 	struct drbd_socket *sock = &first_peer_device(device)->connection->data;
1208 	unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
1209 	struct p_compressed_bm *p = sock->sbuf + header_size;
1210 	int len, err;
1211 
1212 	len = fill_bitmap_rle_bits(device, p,
1213 			DRBD_SOCKET_BUFFER_SIZE - header_size - sizeof(*p), c);
1214 	if (len < 0)
1215 		return -EIO;
1216 
1217 	if (len) {
1218 		dcbp_set_code(p, RLE_VLI_Bits);
1219 		err = __send_command(first_peer_device(device)->connection, device->vnr, sock,
1220 				     P_COMPRESSED_BITMAP, sizeof(*p) + len,
1221 				     NULL, 0);
1222 		c->packets[0]++;
1223 		c->bytes[0] += header_size + sizeof(*p) + len;
1224 
1225 		if (c->bit_offset >= c->bm_bits)
1226 			len = 0; /* DONE */
1227 	} else {
1228 		/* was not compressible.
1229 		 * send a buffer full of plain text bits instead. */
1230 		unsigned int data_size;
1231 		unsigned long num_words;
1232 		unsigned long *p = sock->sbuf + header_size;
1233 
1234 		data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
1235 		num_words = min_t(size_t, data_size / sizeof(*p),
1236 				  c->bm_words - c->word_offset);
1237 		len = num_words * sizeof(*p);
1238 		if (len)
1239 			drbd_bm_get_lel(device, c->word_offset, num_words, p);
1240 		err = __send_command(first_peer_device(device)->connection, device->vnr, sock, P_BITMAP, len, NULL, 0);
1241 		c->word_offset += num_words;
1242 		c->bit_offset = c->word_offset * BITS_PER_LONG;
1243 
1244 		c->packets[1]++;
1245 		c->bytes[1] += header_size + len;
1246 
1247 		if (c->bit_offset > c->bm_bits)
1248 			c->bit_offset = c->bm_bits;
1249 	}
1250 	if (!err) {
1251 		if (len == 0) {
1252 			INFO_bm_xfer_stats(device, "send", c);
1253 			return 0;
1254 		} else
1255 			return 1;
1256 	}
1257 	return -EIO;
1258 }
1259 
1260 /* See the comment at receive_bitmap() */
_drbd_send_bitmap(struct drbd_device * device)1261 static int _drbd_send_bitmap(struct drbd_device *device)
1262 {
1263 	struct bm_xfer_ctx c;
1264 	int err;
1265 
1266 	if (!expect(device->bitmap))
1267 		return false;
1268 
1269 	if (get_ldev(device)) {
1270 		if (drbd_md_test_flag(device->ldev, MDF_FULL_SYNC)) {
1271 			drbd_info(device, "Writing the whole bitmap, MDF_FullSync was set.\n");
1272 			drbd_bm_set_all(device);
1273 			if (drbd_bm_write(device)) {
1274 				/* write_bm did fail! Leave full sync flag set in Meta P_DATA
1275 				 * but otherwise process as per normal - need to tell other
1276 				 * side that a full resync is required! */
1277 				drbd_err(device, "Failed to write bitmap to disk!\n");
1278 			} else {
1279 				drbd_md_clear_flag(device, MDF_FULL_SYNC);
1280 				drbd_md_sync(device);
1281 			}
1282 		}
1283 		put_ldev(device);
1284 	}
1285 
1286 	c = (struct bm_xfer_ctx) {
1287 		.bm_bits = drbd_bm_bits(device),
1288 		.bm_words = drbd_bm_words(device),
1289 	};
1290 
1291 	do {
1292 		err = send_bitmap_rle_or_plain(device, &c);
1293 	} while (err > 0);
1294 
1295 	return err == 0;
1296 }
1297 
drbd_send_bitmap(struct drbd_device * device)1298 int drbd_send_bitmap(struct drbd_device *device)
1299 {
1300 	struct drbd_socket *sock = &first_peer_device(device)->connection->data;
1301 	int err = -1;
1302 
1303 	mutex_lock(&sock->mutex);
1304 	if (sock->socket)
1305 		err = !_drbd_send_bitmap(device);
1306 	mutex_unlock(&sock->mutex);
1307 	return err;
1308 }
1309 
drbd_send_b_ack(struct drbd_connection * connection,u32 barrier_nr,u32 set_size)1310 void drbd_send_b_ack(struct drbd_connection *connection, u32 barrier_nr, u32 set_size)
1311 {
1312 	struct drbd_socket *sock;
1313 	struct p_barrier_ack *p;
1314 
1315 	if (connection->cstate < C_WF_REPORT_PARAMS)
1316 		return;
1317 
1318 	sock = &connection->meta;
1319 	p = conn_prepare_command(connection, sock);
1320 	if (!p)
1321 		return;
1322 	p->barrier = barrier_nr;
1323 	p->set_size = cpu_to_be32(set_size);
1324 	conn_send_command(connection, sock, P_BARRIER_ACK, sizeof(*p), NULL, 0);
1325 }
1326 
1327 /**
1328  * _drbd_send_ack() - Sends an ack packet
1329  * @peer_device:	DRBD peer device.
1330  * @cmd:		Packet command code.
1331  * @sector:		sector, needs to be in big endian byte order
1332  * @blksize:		size in byte, needs to be in big endian byte order
1333  * @block_id:		Id, big endian byte order
1334  */
_drbd_send_ack(struct drbd_peer_device * peer_device,enum drbd_packet cmd,u64 sector,u32 blksize,u64 block_id)1335 static int _drbd_send_ack(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1336 			  u64 sector, u32 blksize, u64 block_id)
1337 {
1338 	struct drbd_socket *sock;
1339 	struct p_block_ack *p;
1340 
1341 	if (peer_device->device->state.conn < C_CONNECTED)
1342 		return -EIO;
1343 
1344 	sock = &peer_device->connection->meta;
1345 	p = drbd_prepare_command(peer_device, sock);
1346 	if (!p)
1347 		return -EIO;
1348 	p->sector = sector;
1349 	p->block_id = block_id;
1350 	p->blksize = blksize;
1351 	p->seq_num = cpu_to_be32(atomic_inc_return(&peer_device->device->packet_seq));
1352 	return drbd_send_command(peer_device, sock, cmd, sizeof(*p), NULL, 0);
1353 }
1354 
1355 /* dp->sector and dp->block_id already/still in network byte order,
1356  * data_size is payload size according to dp->head,
1357  * and may need to be corrected for digest size. */
drbd_send_ack_dp(struct drbd_peer_device * peer_device,enum drbd_packet cmd,struct p_data * dp,int data_size)1358 void drbd_send_ack_dp(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1359 		      struct p_data *dp, int data_size)
1360 {
1361 	if (peer_device->connection->peer_integrity_tfm)
1362 		data_size -= crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1363 	_drbd_send_ack(peer_device, cmd, dp->sector, cpu_to_be32(data_size),
1364 		       dp->block_id);
1365 }
1366 
drbd_send_ack_rp(struct drbd_peer_device * peer_device,enum drbd_packet cmd,struct p_block_req * rp)1367 void drbd_send_ack_rp(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1368 		      struct p_block_req *rp)
1369 {
1370 	_drbd_send_ack(peer_device, cmd, rp->sector, rp->blksize, rp->block_id);
1371 }
1372 
1373 /**
1374  * drbd_send_ack() - Sends an ack packet
1375  * @peer_device:	DRBD peer device
1376  * @cmd:		packet command code
1377  * @peer_req:		peer request
1378  */
drbd_send_ack(struct drbd_peer_device * peer_device,enum drbd_packet cmd,struct drbd_peer_request * peer_req)1379 int drbd_send_ack(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1380 		  struct drbd_peer_request *peer_req)
1381 {
1382 	return _drbd_send_ack(peer_device, cmd,
1383 			      cpu_to_be64(peer_req->i.sector),
1384 			      cpu_to_be32(peer_req->i.size),
1385 			      peer_req->block_id);
1386 }
1387 
1388 /* This function misuses the block_id field to signal if the blocks
1389  * are is sync or not. */
drbd_send_ack_ex(struct drbd_peer_device * peer_device,enum drbd_packet cmd,sector_t sector,int blksize,u64 block_id)1390 int drbd_send_ack_ex(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1391 		     sector_t sector, int blksize, u64 block_id)
1392 {
1393 	return _drbd_send_ack(peer_device, cmd,
1394 			      cpu_to_be64(sector),
1395 			      cpu_to_be32(blksize),
1396 			      cpu_to_be64(block_id));
1397 }
1398 
drbd_send_rs_deallocated(struct drbd_peer_device * peer_device,struct drbd_peer_request * peer_req)1399 int drbd_send_rs_deallocated(struct drbd_peer_device *peer_device,
1400 			     struct drbd_peer_request *peer_req)
1401 {
1402 	struct drbd_socket *sock;
1403 	struct p_block_desc *p;
1404 
1405 	sock = &peer_device->connection->data;
1406 	p = drbd_prepare_command(peer_device, sock);
1407 	if (!p)
1408 		return -EIO;
1409 	p->sector = cpu_to_be64(peer_req->i.sector);
1410 	p->blksize = cpu_to_be32(peer_req->i.size);
1411 	p->pad = 0;
1412 	return drbd_send_command(peer_device, sock, P_RS_DEALLOCATED, sizeof(*p), NULL, 0);
1413 }
1414 
drbd_send_drequest(struct drbd_peer_device * peer_device,int cmd,sector_t sector,int size,u64 block_id)1415 int drbd_send_drequest(struct drbd_peer_device *peer_device, int cmd,
1416 		       sector_t sector, int size, u64 block_id)
1417 {
1418 	struct drbd_socket *sock;
1419 	struct p_block_req *p;
1420 
1421 	sock = &peer_device->connection->data;
1422 	p = drbd_prepare_command(peer_device, sock);
1423 	if (!p)
1424 		return -EIO;
1425 	p->sector = cpu_to_be64(sector);
1426 	p->block_id = block_id;
1427 	p->blksize = cpu_to_be32(size);
1428 	return drbd_send_command(peer_device, sock, cmd, sizeof(*p), NULL, 0);
1429 }
1430 
drbd_send_drequest_csum(struct drbd_peer_device * peer_device,sector_t sector,int size,void * digest,int digest_size,enum drbd_packet cmd)1431 int drbd_send_drequest_csum(struct drbd_peer_device *peer_device, sector_t sector, int size,
1432 			    void *digest, int digest_size, enum drbd_packet cmd)
1433 {
1434 	struct drbd_socket *sock;
1435 	struct p_block_req *p;
1436 
1437 	/* FIXME: Put the digest into the preallocated socket buffer.  */
1438 
1439 	sock = &peer_device->connection->data;
1440 	p = drbd_prepare_command(peer_device, sock);
1441 	if (!p)
1442 		return -EIO;
1443 	p->sector = cpu_to_be64(sector);
1444 	p->block_id = ID_SYNCER /* unused */;
1445 	p->blksize = cpu_to_be32(size);
1446 	return drbd_send_command(peer_device, sock, cmd, sizeof(*p), digest, digest_size);
1447 }
1448 
drbd_send_ov_request(struct drbd_peer_device * peer_device,sector_t sector,int size)1449 int drbd_send_ov_request(struct drbd_peer_device *peer_device, sector_t sector, int size)
1450 {
1451 	struct drbd_socket *sock;
1452 	struct p_block_req *p;
1453 
1454 	sock = &peer_device->connection->data;
1455 	p = drbd_prepare_command(peer_device, sock);
1456 	if (!p)
1457 		return -EIO;
1458 	p->sector = cpu_to_be64(sector);
1459 	p->block_id = ID_SYNCER /* unused */;
1460 	p->blksize = cpu_to_be32(size);
1461 	return drbd_send_command(peer_device, sock, P_OV_REQUEST, sizeof(*p), NULL, 0);
1462 }
1463 
1464 /* called on sndtimeo
1465  * returns false if we should retry,
1466  * true if we think connection is dead
1467  */
we_should_drop_the_connection(struct drbd_connection * connection,struct socket * sock)1468 static int we_should_drop_the_connection(struct drbd_connection *connection, struct socket *sock)
1469 {
1470 	int drop_it;
1471 	/* long elapsed = (long)(jiffies - device->last_received); */
1472 
1473 	drop_it =   connection->meta.socket == sock
1474 		|| !connection->ack_receiver.task
1475 		|| get_t_state(&connection->ack_receiver) != RUNNING
1476 		|| connection->cstate < C_WF_REPORT_PARAMS;
1477 
1478 	if (drop_it)
1479 		return true;
1480 
1481 	drop_it = !--connection->ko_count;
1482 	if (!drop_it) {
1483 		drbd_err(connection, "[%s/%d] sock_sendmsg time expired, ko = %u\n",
1484 			 current->comm, current->pid, connection->ko_count);
1485 		request_ping(connection);
1486 	}
1487 
1488 	return drop_it; /* && (device->state == R_PRIMARY) */;
1489 }
1490 
drbd_update_congested(struct drbd_connection * connection)1491 static void drbd_update_congested(struct drbd_connection *connection)
1492 {
1493 	struct sock *sk = connection->data.socket->sk;
1494 	if (sk->sk_wmem_queued > sk->sk_sndbuf * 4 / 5)
1495 		set_bit(NET_CONGESTED, &connection->flags);
1496 }
1497 
1498 /* The idea of sendpage seems to be to put some kind of reference
1499  * to the page into the skb, and to hand it over to the NIC. In
1500  * this process get_page() gets called.
1501  *
1502  * As soon as the page was really sent over the network put_page()
1503  * gets called by some part of the network layer. [ NIC driver? ]
1504  *
1505  * [ get_page() / put_page() increment/decrement the count. If count
1506  *   reaches 0 the page will be freed. ]
1507  *
1508  * This works nicely with pages from FSs.
1509  * But this means that in protocol A we might signal IO completion too early!
1510  *
1511  * In order not to corrupt data during a resync we must make sure
1512  * that we do not reuse our own buffer pages (EEs) to early, therefore
1513  * we have the net_ee list.
1514  *
1515  * XFS seems to have problems, still, it submits pages with page_count == 0!
1516  * As a workaround, we disable sendpage on pages
1517  * with page_count == 0 or PageSlab.
1518  */
_drbd_no_send_page(struct drbd_peer_device * peer_device,struct page * page,int offset,size_t size,unsigned msg_flags)1519 static int _drbd_no_send_page(struct drbd_peer_device *peer_device, struct page *page,
1520 			      int offset, size_t size, unsigned msg_flags)
1521 {
1522 	struct socket *socket;
1523 	void *addr;
1524 	int err;
1525 
1526 	socket = peer_device->connection->data.socket;
1527 	addr = kmap(page) + offset;
1528 	err = drbd_send_all(peer_device->connection, socket, addr, size, msg_flags);
1529 	kunmap(page);
1530 	if (!err)
1531 		peer_device->device->send_cnt += size >> 9;
1532 	return err;
1533 }
1534 
_drbd_send_page(struct drbd_peer_device * peer_device,struct page * page,int offset,size_t size,unsigned msg_flags)1535 static int _drbd_send_page(struct drbd_peer_device *peer_device, struct page *page,
1536 		    int offset, size_t size, unsigned msg_flags)
1537 {
1538 	struct socket *socket = peer_device->connection->data.socket;
1539 	int len = size;
1540 	int err = -EIO;
1541 
1542 	/* e.g. XFS meta- & log-data is in slab pages, which have a
1543 	 * page_count of 0 and/or have PageSlab() set.
1544 	 * we cannot use send_page for those, as that does get_page();
1545 	 * put_page(); and would cause either a VM_BUG directly, or
1546 	 * __page_cache_release a page that would actually still be referenced
1547 	 * by someone, leading to some obscure delayed Oops somewhere else. */
1548 	if (drbd_disable_sendpage || !sendpage_ok(page))
1549 		return _drbd_no_send_page(peer_device, page, offset, size, msg_flags);
1550 
1551 	msg_flags |= MSG_NOSIGNAL;
1552 	drbd_update_congested(peer_device->connection);
1553 	do {
1554 		int sent;
1555 
1556 		sent = socket->ops->sendpage(socket, page, offset, len, msg_flags);
1557 		if (sent <= 0) {
1558 			if (sent == -EAGAIN) {
1559 				if (we_should_drop_the_connection(peer_device->connection, socket))
1560 					break;
1561 				continue;
1562 			}
1563 			drbd_warn(peer_device->device, "%s: size=%d len=%d sent=%d\n",
1564 			     __func__, (int)size, len, sent);
1565 			if (sent < 0)
1566 				err = sent;
1567 			break;
1568 		}
1569 		len    -= sent;
1570 		offset += sent;
1571 	} while (len > 0 /* THINK && device->cstate >= C_CONNECTED*/);
1572 	clear_bit(NET_CONGESTED, &peer_device->connection->flags);
1573 
1574 	if (len == 0) {
1575 		err = 0;
1576 		peer_device->device->send_cnt += size >> 9;
1577 	}
1578 	return err;
1579 }
1580 
_drbd_send_bio(struct drbd_peer_device * peer_device,struct bio * bio)1581 static int _drbd_send_bio(struct drbd_peer_device *peer_device, struct bio *bio)
1582 {
1583 	struct bio_vec bvec;
1584 	struct bvec_iter iter;
1585 
1586 	/* hint all but last page with MSG_MORE */
1587 	bio_for_each_segment(bvec, bio, iter) {
1588 		int err;
1589 
1590 		err = _drbd_no_send_page(peer_device, bvec.bv_page,
1591 					 bvec.bv_offset, bvec.bv_len,
1592 					 bio_iter_last(bvec, iter)
1593 					 ? 0 : MSG_MORE);
1594 		if (err)
1595 			return err;
1596 		/* REQ_OP_WRITE_SAME has only one segment */
1597 		if (bio_op(bio) == REQ_OP_WRITE_SAME)
1598 			break;
1599 	}
1600 	return 0;
1601 }
1602 
_drbd_send_zc_bio(struct drbd_peer_device * peer_device,struct bio * bio)1603 static int _drbd_send_zc_bio(struct drbd_peer_device *peer_device, struct bio *bio)
1604 {
1605 	struct bio_vec bvec;
1606 	struct bvec_iter iter;
1607 
1608 	/* hint all but last page with MSG_MORE */
1609 	bio_for_each_segment(bvec, bio, iter) {
1610 		int err;
1611 
1612 		err = _drbd_send_page(peer_device, bvec.bv_page,
1613 				      bvec.bv_offset, bvec.bv_len,
1614 				      bio_iter_last(bvec, iter) ? 0 : MSG_MORE);
1615 		if (err)
1616 			return err;
1617 		/* REQ_OP_WRITE_SAME has only one segment */
1618 		if (bio_op(bio) == REQ_OP_WRITE_SAME)
1619 			break;
1620 	}
1621 	return 0;
1622 }
1623 
_drbd_send_zc_ee(struct drbd_peer_device * peer_device,struct drbd_peer_request * peer_req)1624 static int _drbd_send_zc_ee(struct drbd_peer_device *peer_device,
1625 			    struct drbd_peer_request *peer_req)
1626 {
1627 	struct page *page = peer_req->pages;
1628 	unsigned len = peer_req->i.size;
1629 	int err;
1630 
1631 	/* hint all but last page with MSG_MORE */
1632 	page_chain_for_each(page) {
1633 		unsigned l = min_t(unsigned, len, PAGE_SIZE);
1634 
1635 		err = _drbd_send_page(peer_device, page, 0, l,
1636 				      page_chain_next(page) ? MSG_MORE : 0);
1637 		if (err)
1638 			return err;
1639 		len -= l;
1640 	}
1641 	return 0;
1642 }
1643 
bio_flags_to_wire(struct drbd_connection * connection,struct bio * bio)1644 static u32 bio_flags_to_wire(struct drbd_connection *connection,
1645 			     struct bio *bio)
1646 {
1647 	if (connection->agreed_pro_version >= 95)
1648 		return  (bio->bi_opf & REQ_SYNC ? DP_RW_SYNC : 0) |
1649 			(bio->bi_opf & REQ_FUA ? DP_FUA : 0) |
1650 			(bio->bi_opf & REQ_PREFLUSH ? DP_FLUSH : 0) |
1651 			(bio_op(bio) == REQ_OP_WRITE_SAME ? DP_WSAME : 0) |
1652 			(bio_op(bio) == REQ_OP_DISCARD ? DP_DISCARD : 0) |
1653 			(bio_op(bio) == REQ_OP_WRITE_ZEROES ?
1654 			  ((connection->agreed_features & DRBD_FF_WZEROES) ?
1655 			   (DP_ZEROES |(!(bio->bi_opf & REQ_NOUNMAP) ? DP_DISCARD : 0))
1656 			   : DP_DISCARD)
1657 			: 0);
1658 	else
1659 		return bio->bi_opf & REQ_SYNC ? DP_RW_SYNC : 0;
1660 }
1661 
1662 /* Used to send write or TRIM aka REQ_OP_DISCARD requests
1663  * R_PRIMARY -> Peer	(P_DATA, P_TRIM)
1664  */
drbd_send_dblock(struct drbd_peer_device * peer_device,struct drbd_request * req)1665 int drbd_send_dblock(struct drbd_peer_device *peer_device, struct drbd_request *req)
1666 {
1667 	struct drbd_device *device = peer_device->device;
1668 	struct drbd_socket *sock;
1669 	struct p_data *p;
1670 	struct p_wsame *wsame = NULL;
1671 	void *digest_out;
1672 	unsigned int dp_flags = 0;
1673 	int digest_size;
1674 	int err;
1675 
1676 	sock = &peer_device->connection->data;
1677 	p = drbd_prepare_command(peer_device, sock);
1678 	digest_size = peer_device->connection->integrity_tfm ?
1679 		      crypto_shash_digestsize(peer_device->connection->integrity_tfm) : 0;
1680 
1681 	if (!p)
1682 		return -EIO;
1683 	p->sector = cpu_to_be64(req->i.sector);
1684 	p->block_id = (unsigned long)req;
1685 	p->seq_num = cpu_to_be32(atomic_inc_return(&device->packet_seq));
1686 	dp_flags = bio_flags_to_wire(peer_device->connection, req->master_bio);
1687 	if (device->state.conn >= C_SYNC_SOURCE &&
1688 	    device->state.conn <= C_PAUSED_SYNC_T)
1689 		dp_flags |= DP_MAY_SET_IN_SYNC;
1690 	if (peer_device->connection->agreed_pro_version >= 100) {
1691 		if (req->rq_state & RQ_EXP_RECEIVE_ACK)
1692 			dp_flags |= DP_SEND_RECEIVE_ACK;
1693 		/* During resync, request an explicit write ack,
1694 		 * even in protocol != C */
1695 		if (req->rq_state & RQ_EXP_WRITE_ACK
1696 		|| (dp_flags & DP_MAY_SET_IN_SYNC))
1697 			dp_flags |= DP_SEND_WRITE_ACK;
1698 	}
1699 	p->dp_flags = cpu_to_be32(dp_flags);
1700 
1701 	if (dp_flags & (DP_DISCARD|DP_ZEROES)) {
1702 		enum drbd_packet cmd = (dp_flags & DP_ZEROES) ? P_ZEROES : P_TRIM;
1703 		struct p_trim *t = (struct p_trim*)p;
1704 		t->size = cpu_to_be32(req->i.size);
1705 		err = __send_command(peer_device->connection, device->vnr, sock, cmd, sizeof(*t), NULL, 0);
1706 		goto out;
1707 	}
1708 	if (dp_flags & DP_WSAME) {
1709 		/* this will only work if DRBD_FF_WSAME is set AND the
1710 		 * handshake agreed that all nodes and backend devices are
1711 		 * WRITE_SAME capable and agree on logical_block_size */
1712 		wsame = (struct p_wsame*)p;
1713 		digest_out = wsame + 1;
1714 		wsame->size = cpu_to_be32(req->i.size);
1715 	} else
1716 		digest_out = p + 1;
1717 
1718 	/* our digest is still only over the payload.
1719 	 * TRIM does not carry any payload. */
1720 	if (digest_size)
1721 		drbd_csum_bio(peer_device->connection->integrity_tfm, req->master_bio, digest_out);
1722 	if (wsame) {
1723 		err =
1724 		    __send_command(peer_device->connection, device->vnr, sock, P_WSAME,
1725 				   sizeof(*wsame) + digest_size, NULL,
1726 				   bio_iovec(req->master_bio).bv_len);
1727 	} else
1728 		err =
1729 		    __send_command(peer_device->connection, device->vnr, sock, P_DATA,
1730 				   sizeof(*p) + digest_size, NULL, req->i.size);
1731 	if (!err) {
1732 		/* For protocol A, we have to memcpy the payload into
1733 		 * socket buffers, as we may complete right away
1734 		 * as soon as we handed it over to tcp, at which point the data
1735 		 * pages may become invalid.
1736 		 *
1737 		 * For data-integrity enabled, we copy it as well, so we can be
1738 		 * sure that even if the bio pages may still be modified, it
1739 		 * won't change the data on the wire, thus if the digest checks
1740 		 * out ok after sending on this side, but does not fit on the
1741 		 * receiving side, we sure have detected corruption elsewhere.
1742 		 */
1743 		if (!(req->rq_state & (RQ_EXP_RECEIVE_ACK | RQ_EXP_WRITE_ACK)) || digest_size)
1744 			err = _drbd_send_bio(peer_device, req->master_bio);
1745 		else
1746 			err = _drbd_send_zc_bio(peer_device, req->master_bio);
1747 
1748 		/* double check digest, sometimes buffers have been modified in flight. */
1749 		if (digest_size > 0 && digest_size <= 64) {
1750 			/* 64 byte, 512 bit, is the largest digest size
1751 			 * currently supported in kernel crypto. */
1752 			unsigned char digest[64];
1753 			drbd_csum_bio(peer_device->connection->integrity_tfm, req->master_bio, digest);
1754 			if (memcmp(p + 1, digest, digest_size)) {
1755 				drbd_warn(device,
1756 					"Digest mismatch, buffer modified by upper layers during write: %llus +%u\n",
1757 					(unsigned long long)req->i.sector, req->i.size);
1758 			}
1759 		} /* else if (digest_size > 64) {
1760 		     ... Be noisy about digest too large ...
1761 		} */
1762 	}
1763 out:
1764 	mutex_unlock(&sock->mutex);  /* locked by drbd_prepare_command() */
1765 
1766 	return err;
1767 }
1768 
1769 /* answer packet, used to send data back for read requests:
1770  *  Peer       -> (diskless) R_PRIMARY   (P_DATA_REPLY)
1771  *  C_SYNC_SOURCE -> C_SYNC_TARGET         (P_RS_DATA_REPLY)
1772  */
drbd_send_block(struct drbd_peer_device * peer_device,enum drbd_packet cmd,struct drbd_peer_request * peer_req)1773 int drbd_send_block(struct drbd_peer_device *peer_device, enum drbd_packet cmd,
1774 		    struct drbd_peer_request *peer_req)
1775 {
1776 	struct drbd_device *device = peer_device->device;
1777 	struct drbd_socket *sock;
1778 	struct p_data *p;
1779 	int err;
1780 	int digest_size;
1781 
1782 	sock = &peer_device->connection->data;
1783 	p = drbd_prepare_command(peer_device, sock);
1784 
1785 	digest_size = peer_device->connection->integrity_tfm ?
1786 		      crypto_shash_digestsize(peer_device->connection->integrity_tfm) : 0;
1787 
1788 	if (!p)
1789 		return -EIO;
1790 	p->sector = cpu_to_be64(peer_req->i.sector);
1791 	p->block_id = peer_req->block_id;
1792 	p->seq_num = 0;  /* unused */
1793 	p->dp_flags = 0;
1794 	if (digest_size)
1795 		drbd_csum_ee(peer_device->connection->integrity_tfm, peer_req, p + 1);
1796 	err = __send_command(peer_device->connection, device->vnr, sock, cmd, sizeof(*p) + digest_size, NULL, peer_req->i.size);
1797 	if (!err)
1798 		err = _drbd_send_zc_ee(peer_device, peer_req);
1799 	mutex_unlock(&sock->mutex);  /* locked by drbd_prepare_command() */
1800 
1801 	return err;
1802 }
1803 
drbd_send_out_of_sync(struct drbd_peer_device * peer_device,struct drbd_request * req)1804 int drbd_send_out_of_sync(struct drbd_peer_device *peer_device, struct drbd_request *req)
1805 {
1806 	struct drbd_socket *sock;
1807 	struct p_block_desc *p;
1808 
1809 	sock = &peer_device->connection->data;
1810 	p = drbd_prepare_command(peer_device, sock);
1811 	if (!p)
1812 		return -EIO;
1813 	p->sector = cpu_to_be64(req->i.sector);
1814 	p->blksize = cpu_to_be32(req->i.size);
1815 	return drbd_send_command(peer_device, sock, P_OUT_OF_SYNC, sizeof(*p), NULL, 0);
1816 }
1817 
1818 /*
1819   drbd_send distinguishes two cases:
1820 
1821   Packets sent via the data socket "sock"
1822   and packets sent via the meta data socket "msock"
1823 
1824 		    sock                      msock
1825   -----------------+-------------------------+------------------------------
1826   timeout           conf.timeout / 2          conf.timeout / 2
1827   timeout action    send a ping via msock     Abort communication
1828 					      and close all sockets
1829 */
1830 
1831 /*
1832  * you must have down()ed the appropriate [m]sock_mutex elsewhere!
1833  */
drbd_send(struct drbd_connection * connection,struct socket * sock,void * buf,size_t size,unsigned msg_flags)1834 int drbd_send(struct drbd_connection *connection, struct socket *sock,
1835 	      void *buf, size_t size, unsigned msg_flags)
1836 {
1837 	struct kvec iov = {.iov_base = buf, .iov_len = size};
1838 	struct msghdr msg = {.msg_flags = msg_flags | MSG_NOSIGNAL};
1839 	int rv, sent = 0;
1840 
1841 	if (!sock)
1842 		return -EBADR;
1843 
1844 	/* THINK  if (signal_pending) return ... ? */
1845 
1846 	iov_iter_kvec(&msg.msg_iter, WRITE, &iov, 1, size);
1847 
1848 	if (sock == connection->data.socket) {
1849 		rcu_read_lock();
1850 		connection->ko_count = rcu_dereference(connection->net_conf)->ko_count;
1851 		rcu_read_unlock();
1852 		drbd_update_congested(connection);
1853 	}
1854 	do {
1855 		rv = sock_sendmsg(sock, &msg);
1856 		if (rv == -EAGAIN) {
1857 			if (we_should_drop_the_connection(connection, sock))
1858 				break;
1859 			else
1860 				continue;
1861 		}
1862 		if (rv == -EINTR) {
1863 			flush_signals(current);
1864 			rv = 0;
1865 		}
1866 		if (rv < 0)
1867 			break;
1868 		sent += rv;
1869 	} while (sent < size);
1870 
1871 	if (sock == connection->data.socket)
1872 		clear_bit(NET_CONGESTED, &connection->flags);
1873 
1874 	if (rv <= 0) {
1875 		if (rv != -EAGAIN) {
1876 			drbd_err(connection, "%s_sendmsg returned %d\n",
1877 				 sock == connection->meta.socket ? "msock" : "sock",
1878 				 rv);
1879 			conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
1880 		} else
1881 			conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
1882 	}
1883 
1884 	return sent;
1885 }
1886 
1887 /*
1888  * drbd_send_all  -  Send an entire buffer
1889  *
1890  * Returns 0 upon success and a negative error value otherwise.
1891  */
drbd_send_all(struct drbd_connection * connection,struct socket * sock,void * buffer,size_t size,unsigned msg_flags)1892 int drbd_send_all(struct drbd_connection *connection, struct socket *sock, void *buffer,
1893 		  size_t size, unsigned msg_flags)
1894 {
1895 	int err;
1896 
1897 	err = drbd_send(connection, sock, buffer, size, msg_flags);
1898 	if (err < 0)
1899 		return err;
1900 	if (err != size)
1901 		return -EIO;
1902 	return 0;
1903 }
1904 
drbd_open(struct block_device * bdev,fmode_t mode)1905 static int drbd_open(struct block_device *bdev, fmode_t mode)
1906 {
1907 	struct drbd_device *device = bdev->bd_disk->private_data;
1908 	unsigned long flags;
1909 	int rv = 0;
1910 
1911 	mutex_lock(&drbd_main_mutex);
1912 	spin_lock_irqsave(&device->resource->req_lock, flags);
1913 	/* to have a stable device->state.role
1914 	 * and no race with updating open_cnt */
1915 
1916 	if (device->state.role != R_PRIMARY) {
1917 		if (mode & FMODE_WRITE)
1918 			rv = -EROFS;
1919 		else if (!drbd_allow_oos)
1920 			rv = -EMEDIUMTYPE;
1921 	}
1922 
1923 	if (!rv)
1924 		device->open_cnt++;
1925 	spin_unlock_irqrestore(&device->resource->req_lock, flags);
1926 	mutex_unlock(&drbd_main_mutex);
1927 
1928 	return rv;
1929 }
1930 
drbd_release(struct gendisk * gd,fmode_t mode)1931 static void drbd_release(struct gendisk *gd, fmode_t mode)
1932 {
1933 	struct drbd_device *device = gd->private_data;
1934 	mutex_lock(&drbd_main_mutex);
1935 	device->open_cnt--;
1936 	mutex_unlock(&drbd_main_mutex);
1937 }
1938 
1939 /* need to hold resource->req_lock */
drbd_queue_unplug(struct drbd_device * device)1940 void drbd_queue_unplug(struct drbd_device *device)
1941 {
1942 	if (device->state.pdsk >= D_INCONSISTENT && device->state.conn >= C_CONNECTED) {
1943 		D_ASSERT(device, device->state.role == R_PRIMARY);
1944 		if (test_and_clear_bit(UNPLUG_REMOTE, &device->flags)) {
1945 			drbd_queue_work_if_unqueued(
1946 				&first_peer_device(device)->connection->sender_work,
1947 				&device->unplug_work);
1948 		}
1949 	}
1950 }
1951 
drbd_set_defaults(struct drbd_device * device)1952 static void drbd_set_defaults(struct drbd_device *device)
1953 {
1954 	/* Beware! The actual layout differs
1955 	 * between big endian and little endian */
1956 	device->state = (union drbd_dev_state) {
1957 		{ .role = R_SECONDARY,
1958 		  .peer = R_UNKNOWN,
1959 		  .conn = C_STANDALONE,
1960 		  .disk = D_DISKLESS,
1961 		  .pdsk = D_UNKNOWN,
1962 		} };
1963 }
1964 
drbd_init_set_defaults(struct drbd_device * device)1965 void drbd_init_set_defaults(struct drbd_device *device)
1966 {
1967 	/* the memset(,0,) did most of this.
1968 	 * note: only assignments, no allocation in here */
1969 
1970 	drbd_set_defaults(device);
1971 
1972 	atomic_set(&device->ap_bio_cnt, 0);
1973 	atomic_set(&device->ap_actlog_cnt, 0);
1974 	atomic_set(&device->ap_pending_cnt, 0);
1975 	atomic_set(&device->rs_pending_cnt, 0);
1976 	atomic_set(&device->unacked_cnt, 0);
1977 	atomic_set(&device->local_cnt, 0);
1978 	atomic_set(&device->pp_in_use_by_net, 0);
1979 	atomic_set(&device->rs_sect_in, 0);
1980 	atomic_set(&device->rs_sect_ev, 0);
1981 	atomic_set(&device->ap_in_flight, 0);
1982 	atomic_set(&device->md_io.in_use, 0);
1983 
1984 	mutex_init(&device->own_state_mutex);
1985 	device->state_mutex = &device->own_state_mutex;
1986 
1987 	spin_lock_init(&device->al_lock);
1988 	spin_lock_init(&device->peer_seq_lock);
1989 
1990 	INIT_LIST_HEAD(&device->active_ee);
1991 	INIT_LIST_HEAD(&device->sync_ee);
1992 	INIT_LIST_HEAD(&device->done_ee);
1993 	INIT_LIST_HEAD(&device->read_ee);
1994 	INIT_LIST_HEAD(&device->net_ee);
1995 	INIT_LIST_HEAD(&device->resync_reads);
1996 	INIT_LIST_HEAD(&device->resync_work.list);
1997 	INIT_LIST_HEAD(&device->unplug_work.list);
1998 	INIT_LIST_HEAD(&device->bm_io_work.w.list);
1999 	INIT_LIST_HEAD(&device->pending_master_completion[0]);
2000 	INIT_LIST_HEAD(&device->pending_master_completion[1]);
2001 	INIT_LIST_HEAD(&device->pending_completion[0]);
2002 	INIT_LIST_HEAD(&device->pending_completion[1]);
2003 
2004 	device->resync_work.cb  = w_resync_timer;
2005 	device->unplug_work.cb  = w_send_write_hint;
2006 	device->bm_io_work.w.cb = w_bitmap_io;
2007 
2008 	timer_setup(&device->resync_timer, resync_timer_fn, 0);
2009 	timer_setup(&device->md_sync_timer, md_sync_timer_fn, 0);
2010 	timer_setup(&device->start_resync_timer, start_resync_timer_fn, 0);
2011 	timer_setup(&device->request_timer, request_timer_fn, 0);
2012 
2013 	init_waitqueue_head(&device->misc_wait);
2014 	init_waitqueue_head(&device->state_wait);
2015 	init_waitqueue_head(&device->ee_wait);
2016 	init_waitqueue_head(&device->al_wait);
2017 	init_waitqueue_head(&device->seq_wait);
2018 
2019 	device->resync_wenr = LC_FREE;
2020 	device->peer_max_bio_size = DRBD_MAX_BIO_SIZE_SAFE;
2021 	device->local_max_bio_size = DRBD_MAX_BIO_SIZE_SAFE;
2022 }
2023 
drbd_set_my_capacity(struct drbd_device * device,sector_t size)2024 void drbd_set_my_capacity(struct drbd_device *device, sector_t size)
2025 {
2026 	char ppb[10];
2027 
2028 	set_capacity_and_notify(device->vdisk, size);
2029 
2030 	drbd_info(device, "size = %s (%llu KB)\n",
2031 		ppsize(ppb, size>>1), (unsigned long long)size>>1);
2032 }
2033 
drbd_device_cleanup(struct drbd_device * device)2034 void drbd_device_cleanup(struct drbd_device *device)
2035 {
2036 	int i;
2037 	if (first_peer_device(device)->connection->receiver.t_state != NONE)
2038 		drbd_err(device, "ASSERT FAILED: receiver t_state == %d expected 0.\n",
2039 				first_peer_device(device)->connection->receiver.t_state);
2040 
2041 	device->al_writ_cnt  =
2042 	device->bm_writ_cnt  =
2043 	device->read_cnt     =
2044 	device->recv_cnt     =
2045 	device->send_cnt     =
2046 	device->writ_cnt     =
2047 	device->p_size       =
2048 	device->rs_start     =
2049 	device->rs_total     =
2050 	device->rs_failed    = 0;
2051 	device->rs_last_events = 0;
2052 	device->rs_last_sect_ev = 0;
2053 	for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2054 		device->rs_mark_left[i] = 0;
2055 		device->rs_mark_time[i] = 0;
2056 	}
2057 	D_ASSERT(device, first_peer_device(device)->connection->net_conf == NULL);
2058 
2059 	set_capacity_and_notify(device->vdisk, 0);
2060 	if (device->bitmap) {
2061 		/* maybe never allocated. */
2062 		drbd_bm_resize(device, 0, 1);
2063 		drbd_bm_cleanup(device);
2064 	}
2065 
2066 	drbd_backing_dev_free(device, device->ldev);
2067 	device->ldev = NULL;
2068 
2069 	clear_bit(AL_SUSPENDED, &device->flags);
2070 
2071 	D_ASSERT(device, list_empty(&device->active_ee));
2072 	D_ASSERT(device, list_empty(&device->sync_ee));
2073 	D_ASSERT(device, list_empty(&device->done_ee));
2074 	D_ASSERT(device, list_empty(&device->read_ee));
2075 	D_ASSERT(device, list_empty(&device->net_ee));
2076 	D_ASSERT(device, list_empty(&device->resync_reads));
2077 	D_ASSERT(device, list_empty(&first_peer_device(device)->connection->sender_work.q));
2078 	D_ASSERT(device, list_empty(&device->resync_work.list));
2079 	D_ASSERT(device, list_empty(&device->unplug_work.list));
2080 
2081 	drbd_set_defaults(device);
2082 }
2083 
2084 
drbd_destroy_mempools(void)2085 static void drbd_destroy_mempools(void)
2086 {
2087 	struct page *page;
2088 
2089 	while (drbd_pp_pool) {
2090 		page = drbd_pp_pool;
2091 		drbd_pp_pool = (struct page *)page_private(page);
2092 		__free_page(page);
2093 		drbd_pp_vacant--;
2094 	}
2095 
2096 	/* D_ASSERT(device, atomic_read(&drbd_pp_vacant)==0); */
2097 
2098 	bioset_exit(&drbd_io_bio_set);
2099 	bioset_exit(&drbd_md_io_bio_set);
2100 	mempool_exit(&drbd_md_io_page_pool);
2101 	mempool_exit(&drbd_ee_mempool);
2102 	mempool_exit(&drbd_request_mempool);
2103 	kmem_cache_destroy(drbd_ee_cache);
2104 	kmem_cache_destroy(drbd_request_cache);
2105 	kmem_cache_destroy(drbd_bm_ext_cache);
2106 	kmem_cache_destroy(drbd_al_ext_cache);
2107 
2108 	drbd_ee_cache        = NULL;
2109 	drbd_request_cache   = NULL;
2110 	drbd_bm_ext_cache    = NULL;
2111 	drbd_al_ext_cache    = NULL;
2112 
2113 	return;
2114 }
2115 
drbd_create_mempools(void)2116 static int drbd_create_mempools(void)
2117 {
2118 	struct page *page;
2119 	const int number = (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * drbd_minor_count;
2120 	int i, ret;
2121 
2122 	/* caches */
2123 	drbd_request_cache = kmem_cache_create(
2124 		"drbd_req", sizeof(struct drbd_request), 0, 0, NULL);
2125 	if (drbd_request_cache == NULL)
2126 		goto Enomem;
2127 
2128 	drbd_ee_cache = kmem_cache_create(
2129 		"drbd_ee", sizeof(struct drbd_peer_request), 0, 0, NULL);
2130 	if (drbd_ee_cache == NULL)
2131 		goto Enomem;
2132 
2133 	drbd_bm_ext_cache = kmem_cache_create(
2134 		"drbd_bm", sizeof(struct bm_extent), 0, 0, NULL);
2135 	if (drbd_bm_ext_cache == NULL)
2136 		goto Enomem;
2137 
2138 	drbd_al_ext_cache = kmem_cache_create(
2139 		"drbd_al", sizeof(struct lc_element), 0, 0, NULL);
2140 	if (drbd_al_ext_cache == NULL)
2141 		goto Enomem;
2142 
2143 	/* mempools */
2144 	ret = bioset_init(&drbd_io_bio_set, BIO_POOL_SIZE, 0, 0);
2145 	if (ret)
2146 		goto Enomem;
2147 
2148 	ret = bioset_init(&drbd_md_io_bio_set, DRBD_MIN_POOL_PAGES, 0,
2149 			  BIOSET_NEED_BVECS);
2150 	if (ret)
2151 		goto Enomem;
2152 
2153 	ret = mempool_init_page_pool(&drbd_md_io_page_pool, DRBD_MIN_POOL_PAGES, 0);
2154 	if (ret)
2155 		goto Enomem;
2156 
2157 	ret = mempool_init_slab_pool(&drbd_request_mempool, number,
2158 				     drbd_request_cache);
2159 	if (ret)
2160 		goto Enomem;
2161 
2162 	ret = mempool_init_slab_pool(&drbd_ee_mempool, number, drbd_ee_cache);
2163 	if (ret)
2164 		goto Enomem;
2165 
2166 	for (i = 0; i < number; i++) {
2167 		page = alloc_page(GFP_HIGHUSER);
2168 		if (!page)
2169 			goto Enomem;
2170 		set_page_private(page, (unsigned long)drbd_pp_pool);
2171 		drbd_pp_pool = page;
2172 	}
2173 	drbd_pp_vacant = number;
2174 
2175 	return 0;
2176 
2177 Enomem:
2178 	drbd_destroy_mempools(); /* in case we allocated some */
2179 	return -ENOMEM;
2180 }
2181 
drbd_release_all_peer_reqs(struct drbd_device * device)2182 static void drbd_release_all_peer_reqs(struct drbd_device *device)
2183 {
2184 	int rr;
2185 
2186 	rr = drbd_free_peer_reqs(device, &device->active_ee);
2187 	if (rr)
2188 		drbd_err(device, "%d EEs in active list found!\n", rr);
2189 
2190 	rr = drbd_free_peer_reqs(device, &device->sync_ee);
2191 	if (rr)
2192 		drbd_err(device, "%d EEs in sync list found!\n", rr);
2193 
2194 	rr = drbd_free_peer_reqs(device, &device->read_ee);
2195 	if (rr)
2196 		drbd_err(device, "%d EEs in read list found!\n", rr);
2197 
2198 	rr = drbd_free_peer_reqs(device, &device->done_ee);
2199 	if (rr)
2200 		drbd_err(device, "%d EEs in done list found!\n", rr);
2201 
2202 	rr = drbd_free_peer_reqs(device, &device->net_ee);
2203 	if (rr)
2204 		drbd_err(device, "%d EEs in net list found!\n", rr);
2205 }
2206 
2207 /* caution. no locking. */
drbd_destroy_device(struct kref * kref)2208 void drbd_destroy_device(struct kref *kref)
2209 {
2210 	struct drbd_device *device = container_of(kref, struct drbd_device, kref);
2211 	struct drbd_resource *resource = device->resource;
2212 	struct drbd_peer_device *peer_device, *tmp_peer_device;
2213 
2214 	del_timer_sync(&device->request_timer);
2215 
2216 	/* paranoia asserts */
2217 	D_ASSERT(device, device->open_cnt == 0);
2218 	/* end paranoia asserts */
2219 
2220 	/* cleanup stuff that may have been allocated during
2221 	 * device (re-)configuration or state changes */
2222 
2223 	drbd_backing_dev_free(device, device->ldev);
2224 	device->ldev = NULL;
2225 
2226 	drbd_release_all_peer_reqs(device);
2227 
2228 	lc_destroy(device->act_log);
2229 	lc_destroy(device->resync);
2230 
2231 	kfree(device->p_uuid);
2232 	/* device->p_uuid = NULL; */
2233 
2234 	if (device->bitmap) /* should no longer be there. */
2235 		drbd_bm_cleanup(device);
2236 	__free_page(device->md_io.page);
2237 	blk_cleanup_disk(device->vdisk);
2238 	kfree(device->rs_plan_s);
2239 
2240 	/* not for_each_connection(connection, resource):
2241 	 * those may have been cleaned up and disassociated already.
2242 	 */
2243 	for_each_peer_device_safe(peer_device, tmp_peer_device, device) {
2244 		kref_put(&peer_device->connection->kref, drbd_destroy_connection);
2245 		kfree(peer_device);
2246 	}
2247 	if (device->submit.wq)
2248 		destroy_workqueue(device->submit.wq);
2249 	kfree(device);
2250 	kref_put(&resource->kref, drbd_destroy_resource);
2251 }
2252 
2253 /* One global retry thread, if we need to push back some bio and have it
2254  * reinserted through our make request function.
2255  */
2256 static struct retry_worker {
2257 	struct workqueue_struct *wq;
2258 	struct work_struct worker;
2259 
2260 	spinlock_t lock;
2261 	struct list_head writes;
2262 } retry;
2263 
do_retry(struct work_struct * ws)2264 static void do_retry(struct work_struct *ws)
2265 {
2266 	struct retry_worker *retry = container_of(ws, struct retry_worker, worker);
2267 	LIST_HEAD(writes);
2268 	struct drbd_request *req, *tmp;
2269 
2270 	spin_lock_irq(&retry->lock);
2271 	list_splice_init(&retry->writes, &writes);
2272 	spin_unlock_irq(&retry->lock);
2273 
2274 	list_for_each_entry_safe(req, tmp, &writes, tl_requests) {
2275 		struct drbd_device *device = req->device;
2276 		struct bio *bio = req->master_bio;
2277 		bool expected;
2278 
2279 		expected =
2280 			expect(atomic_read(&req->completion_ref) == 0) &&
2281 			expect(req->rq_state & RQ_POSTPONED) &&
2282 			expect((req->rq_state & RQ_LOCAL_PENDING) == 0 ||
2283 				(req->rq_state & RQ_LOCAL_ABORTED) != 0);
2284 
2285 		if (!expected)
2286 			drbd_err(device, "req=%p completion_ref=%d rq_state=%x\n",
2287 				req, atomic_read(&req->completion_ref),
2288 				req->rq_state);
2289 
2290 		/* We still need to put one kref associated with the
2291 		 * "completion_ref" going zero in the code path that queued it
2292 		 * here.  The request object may still be referenced by a
2293 		 * frozen local req->private_bio, in case we force-detached.
2294 		 */
2295 		kref_put(&req->kref, drbd_req_destroy);
2296 
2297 		/* A single suspended or otherwise blocking device may stall
2298 		 * all others as well.  Fortunately, this code path is to
2299 		 * recover from a situation that "should not happen":
2300 		 * concurrent writes in multi-primary setup.
2301 		 * In a "normal" lifecycle, this workqueue is supposed to be
2302 		 * destroyed without ever doing anything.
2303 		 * If it turns out to be an issue anyways, we can do per
2304 		 * resource (replication group) or per device (minor) retry
2305 		 * workqueues instead.
2306 		 */
2307 
2308 		/* We are not just doing submit_bio_noacct(),
2309 		 * as we want to keep the start_time information. */
2310 		inc_ap_bio(device);
2311 		__drbd_make_request(device, bio);
2312 	}
2313 }
2314 
2315 /* called via drbd_req_put_completion_ref(),
2316  * holds resource->req_lock */
drbd_restart_request(struct drbd_request * req)2317 void drbd_restart_request(struct drbd_request *req)
2318 {
2319 	unsigned long flags;
2320 	spin_lock_irqsave(&retry.lock, flags);
2321 	list_move_tail(&req->tl_requests, &retry.writes);
2322 	spin_unlock_irqrestore(&retry.lock, flags);
2323 
2324 	/* Drop the extra reference that would otherwise
2325 	 * have been dropped by complete_master_bio.
2326 	 * do_retry() needs to grab a new one. */
2327 	dec_ap_bio(req->device);
2328 
2329 	queue_work(retry.wq, &retry.worker);
2330 }
2331 
drbd_destroy_resource(struct kref * kref)2332 void drbd_destroy_resource(struct kref *kref)
2333 {
2334 	struct drbd_resource *resource =
2335 		container_of(kref, struct drbd_resource, kref);
2336 
2337 	idr_destroy(&resource->devices);
2338 	free_cpumask_var(resource->cpu_mask);
2339 	kfree(resource->name);
2340 	kfree(resource);
2341 }
2342 
drbd_free_resource(struct drbd_resource * resource)2343 void drbd_free_resource(struct drbd_resource *resource)
2344 {
2345 	struct drbd_connection *connection, *tmp;
2346 
2347 	for_each_connection_safe(connection, tmp, resource) {
2348 		list_del(&connection->connections);
2349 		drbd_debugfs_connection_cleanup(connection);
2350 		kref_put(&connection->kref, drbd_destroy_connection);
2351 	}
2352 	drbd_debugfs_resource_cleanup(resource);
2353 	kref_put(&resource->kref, drbd_destroy_resource);
2354 }
2355 
drbd_cleanup(void)2356 static void drbd_cleanup(void)
2357 {
2358 	unsigned int i;
2359 	struct drbd_device *device;
2360 	struct drbd_resource *resource, *tmp;
2361 
2362 	/* first remove proc,
2363 	 * drbdsetup uses it's presence to detect
2364 	 * whether DRBD is loaded.
2365 	 * If we would get stuck in proc removal,
2366 	 * but have netlink already deregistered,
2367 	 * some drbdsetup commands may wait forever
2368 	 * for an answer.
2369 	 */
2370 	if (drbd_proc)
2371 		remove_proc_entry("drbd", NULL);
2372 
2373 	if (retry.wq)
2374 		destroy_workqueue(retry.wq);
2375 
2376 	drbd_genl_unregister();
2377 
2378 	idr_for_each_entry(&drbd_devices, device, i)
2379 		drbd_delete_device(device);
2380 
2381 	/* not _rcu since, no other updater anymore. Genl already unregistered */
2382 	for_each_resource_safe(resource, tmp, &drbd_resources) {
2383 		list_del(&resource->resources);
2384 		drbd_free_resource(resource);
2385 	}
2386 
2387 	drbd_debugfs_cleanup();
2388 
2389 	drbd_destroy_mempools();
2390 	unregister_blkdev(DRBD_MAJOR, "drbd");
2391 
2392 	idr_destroy(&drbd_devices);
2393 
2394 	pr_info("module cleanup done.\n");
2395 }
2396 
drbd_init_workqueue(struct drbd_work_queue * wq)2397 static void drbd_init_workqueue(struct drbd_work_queue* wq)
2398 {
2399 	spin_lock_init(&wq->q_lock);
2400 	INIT_LIST_HEAD(&wq->q);
2401 	init_waitqueue_head(&wq->q_wait);
2402 }
2403 
2404 struct completion_work {
2405 	struct drbd_work w;
2406 	struct completion done;
2407 };
2408 
w_complete(struct drbd_work * w,int cancel)2409 static int w_complete(struct drbd_work *w, int cancel)
2410 {
2411 	struct completion_work *completion_work =
2412 		container_of(w, struct completion_work, w);
2413 
2414 	complete(&completion_work->done);
2415 	return 0;
2416 }
2417 
drbd_flush_workqueue(struct drbd_work_queue * work_queue)2418 void drbd_flush_workqueue(struct drbd_work_queue *work_queue)
2419 {
2420 	struct completion_work completion_work;
2421 
2422 	completion_work.w.cb = w_complete;
2423 	init_completion(&completion_work.done);
2424 	drbd_queue_work(work_queue, &completion_work.w);
2425 	wait_for_completion(&completion_work.done);
2426 }
2427 
drbd_find_resource(const char * name)2428 struct drbd_resource *drbd_find_resource(const char *name)
2429 {
2430 	struct drbd_resource *resource;
2431 
2432 	if (!name || !name[0])
2433 		return NULL;
2434 
2435 	rcu_read_lock();
2436 	for_each_resource_rcu(resource, &drbd_resources) {
2437 		if (!strcmp(resource->name, name)) {
2438 			kref_get(&resource->kref);
2439 			goto found;
2440 		}
2441 	}
2442 	resource = NULL;
2443 found:
2444 	rcu_read_unlock();
2445 	return resource;
2446 }
2447 
conn_get_by_addrs(void * my_addr,int my_addr_len,void * peer_addr,int peer_addr_len)2448 struct drbd_connection *conn_get_by_addrs(void *my_addr, int my_addr_len,
2449 				     void *peer_addr, int peer_addr_len)
2450 {
2451 	struct drbd_resource *resource;
2452 	struct drbd_connection *connection;
2453 
2454 	rcu_read_lock();
2455 	for_each_resource_rcu(resource, &drbd_resources) {
2456 		for_each_connection_rcu(connection, resource) {
2457 			if (connection->my_addr_len == my_addr_len &&
2458 			    connection->peer_addr_len == peer_addr_len &&
2459 			    !memcmp(&connection->my_addr, my_addr, my_addr_len) &&
2460 			    !memcmp(&connection->peer_addr, peer_addr, peer_addr_len)) {
2461 				kref_get(&connection->kref);
2462 				goto found;
2463 			}
2464 		}
2465 	}
2466 	connection = NULL;
2467 found:
2468 	rcu_read_unlock();
2469 	return connection;
2470 }
2471 
drbd_alloc_socket(struct drbd_socket * socket)2472 static int drbd_alloc_socket(struct drbd_socket *socket)
2473 {
2474 	socket->rbuf = (void *) __get_free_page(GFP_KERNEL);
2475 	if (!socket->rbuf)
2476 		return -ENOMEM;
2477 	socket->sbuf = (void *) __get_free_page(GFP_KERNEL);
2478 	if (!socket->sbuf)
2479 		return -ENOMEM;
2480 	return 0;
2481 }
2482 
drbd_free_socket(struct drbd_socket * socket)2483 static void drbd_free_socket(struct drbd_socket *socket)
2484 {
2485 	free_page((unsigned long) socket->sbuf);
2486 	free_page((unsigned long) socket->rbuf);
2487 }
2488 
conn_free_crypto(struct drbd_connection * connection)2489 void conn_free_crypto(struct drbd_connection *connection)
2490 {
2491 	drbd_free_sock(connection);
2492 
2493 	crypto_free_shash(connection->csums_tfm);
2494 	crypto_free_shash(connection->verify_tfm);
2495 	crypto_free_shash(connection->cram_hmac_tfm);
2496 	crypto_free_shash(connection->integrity_tfm);
2497 	crypto_free_shash(connection->peer_integrity_tfm);
2498 	kfree(connection->int_dig_in);
2499 	kfree(connection->int_dig_vv);
2500 
2501 	connection->csums_tfm = NULL;
2502 	connection->verify_tfm = NULL;
2503 	connection->cram_hmac_tfm = NULL;
2504 	connection->integrity_tfm = NULL;
2505 	connection->peer_integrity_tfm = NULL;
2506 	connection->int_dig_in = NULL;
2507 	connection->int_dig_vv = NULL;
2508 }
2509 
set_resource_options(struct drbd_resource * resource,struct res_opts * res_opts)2510 int set_resource_options(struct drbd_resource *resource, struct res_opts *res_opts)
2511 {
2512 	struct drbd_connection *connection;
2513 	cpumask_var_t new_cpu_mask;
2514 	int err;
2515 
2516 	if (!zalloc_cpumask_var(&new_cpu_mask, GFP_KERNEL))
2517 		return -ENOMEM;
2518 
2519 	/* silently ignore cpu mask on UP kernel */
2520 	if (nr_cpu_ids > 1 && res_opts->cpu_mask[0] != 0) {
2521 		err = bitmap_parse(res_opts->cpu_mask, DRBD_CPU_MASK_SIZE,
2522 				   cpumask_bits(new_cpu_mask), nr_cpu_ids);
2523 		if (err == -EOVERFLOW) {
2524 			/* So what. mask it out. */
2525 			cpumask_var_t tmp_cpu_mask;
2526 			if (zalloc_cpumask_var(&tmp_cpu_mask, GFP_KERNEL)) {
2527 				cpumask_setall(tmp_cpu_mask);
2528 				cpumask_and(new_cpu_mask, new_cpu_mask, tmp_cpu_mask);
2529 				drbd_warn(resource, "Overflow in bitmap_parse(%.12s%s), truncating to %u bits\n",
2530 					res_opts->cpu_mask,
2531 					strlen(res_opts->cpu_mask) > 12 ? "..." : "",
2532 					nr_cpu_ids);
2533 				free_cpumask_var(tmp_cpu_mask);
2534 				err = 0;
2535 			}
2536 		}
2537 		if (err) {
2538 			drbd_warn(resource, "bitmap_parse() failed with %d\n", err);
2539 			/* retcode = ERR_CPU_MASK_PARSE; */
2540 			goto fail;
2541 		}
2542 	}
2543 	resource->res_opts = *res_opts;
2544 	if (cpumask_empty(new_cpu_mask))
2545 		drbd_calc_cpu_mask(&new_cpu_mask);
2546 	if (!cpumask_equal(resource->cpu_mask, new_cpu_mask)) {
2547 		cpumask_copy(resource->cpu_mask, new_cpu_mask);
2548 		for_each_connection_rcu(connection, resource) {
2549 			connection->receiver.reset_cpu_mask = 1;
2550 			connection->ack_receiver.reset_cpu_mask = 1;
2551 			connection->worker.reset_cpu_mask = 1;
2552 		}
2553 	}
2554 	err = 0;
2555 
2556 fail:
2557 	free_cpumask_var(new_cpu_mask);
2558 	return err;
2559 
2560 }
2561 
drbd_create_resource(const char * name)2562 struct drbd_resource *drbd_create_resource(const char *name)
2563 {
2564 	struct drbd_resource *resource;
2565 
2566 	resource = kzalloc(sizeof(struct drbd_resource), GFP_KERNEL);
2567 	if (!resource)
2568 		goto fail;
2569 	resource->name = kstrdup(name, GFP_KERNEL);
2570 	if (!resource->name)
2571 		goto fail_free_resource;
2572 	if (!zalloc_cpumask_var(&resource->cpu_mask, GFP_KERNEL))
2573 		goto fail_free_name;
2574 	kref_init(&resource->kref);
2575 	idr_init(&resource->devices);
2576 	INIT_LIST_HEAD(&resource->connections);
2577 	resource->write_ordering = WO_BDEV_FLUSH;
2578 	list_add_tail_rcu(&resource->resources, &drbd_resources);
2579 	mutex_init(&resource->conf_update);
2580 	mutex_init(&resource->adm_mutex);
2581 	spin_lock_init(&resource->req_lock);
2582 	drbd_debugfs_resource_add(resource);
2583 	return resource;
2584 
2585 fail_free_name:
2586 	kfree(resource->name);
2587 fail_free_resource:
2588 	kfree(resource);
2589 fail:
2590 	return NULL;
2591 }
2592 
2593 /* caller must be under adm_mutex */
conn_create(const char * name,struct res_opts * res_opts)2594 struct drbd_connection *conn_create(const char *name, struct res_opts *res_opts)
2595 {
2596 	struct drbd_resource *resource;
2597 	struct drbd_connection *connection;
2598 
2599 	connection = kzalloc(sizeof(struct drbd_connection), GFP_KERNEL);
2600 	if (!connection)
2601 		return NULL;
2602 
2603 	if (drbd_alloc_socket(&connection->data))
2604 		goto fail;
2605 	if (drbd_alloc_socket(&connection->meta))
2606 		goto fail;
2607 
2608 	connection->current_epoch = kzalloc(sizeof(struct drbd_epoch), GFP_KERNEL);
2609 	if (!connection->current_epoch)
2610 		goto fail;
2611 
2612 	INIT_LIST_HEAD(&connection->transfer_log);
2613 
2614 	INIT_LIST_HEAD(&connection->current_epoch->list);
2615 	connection->epochs = 1;
2616 	spin_lock_init(&connection->epoch_lock);
2617 
2618 	connection->send.seen_any_write_yet = false;
2619 	connection->send.current_epoch_nr = 0;
2620 	connection->send.current_epoch_writes = 0;
2621 
2622 	resource = drbd_create_resource(name);
2623 	if (!resource)
2624 		goto fail;
2625 
2626 	connection->cstate = C_STANDALONE;
2627 	mutex_init(&connection->cstate_mutex);
2628 	init_waitqueue_head(&connection->ping_wait);
2629 	idr_init(&connection->peer_devices);
2630 
2631 	drbd_init_workqueue(&connection->sender_work);
2632 	mutex_init(&connection->data.mutex);
2633 	mutex_init(&connection->meta.mutex);
2634 
2635 	drbd_thread_init(resource, &connection->receiver, drbd_receiver, "receiver");
2636 	connection->receiver.connection = connection;
2637 	drbd_thread_init(resource, &connection->worker, drbd_worker, "worker");
2638 	connection->worker.connection = connection;
2639 	drbd_thread_init(resource, &connection->ack_receiver, drbd_ack_receiver, "ack_recv");
2640 	connection->ack_receiver.connection = connection;
2641 
2642 	kref_init(&connection->kref);
2643 
2644 	connection->resource = resource;
2645 
2646 	if (set_resource_options(resource, res_opts))
2647 		goto fail_resource;
2648 
2649 	kref_get(&resource->kref);
2650 	list_add_tail_rcu(&connection->connections, &resource->connections);
2651 	drbd_debugfs_connection_add(connection);
2652 	return connection;
2653 
2654 fail_resource:
2655 	list_del(&resource->resources);
2656 	drbd_free_resource(resource);
2657 fail:
2658 	kfree(connection->current_epoch);
2659 	drbd_free_socket(&connection->meta);
2660 	drbd_free_socket(&connection->data);
2661 	kfree(connection);
2662 	return NULL;
2663 }
2664 
drbd_destroy_connection(struct kref * kref)2665 void drbd_destroy_connection(struct kref *kref)
2666 {
2667 	struct drbd_connection *connection = container_of(kref, struct drbd_connection, kref);
2668 	struct drbd_resource *resource = connection->resource;
2669 
2670 	if (atomic_read(&connection->current_epoch->epoch_size) !=  0)
2671 		drbd_err(connection, "epoch_size:%d\n", atomic_read(&connection->current_epoch->epoch_size));
2672 	kfree(connection->current_epoch);
2673 
2674 	idr_destroy(&connection->peer_devices);
2675 
2676 	drbd_free_socket(&connection->meta);
2677 	drbd_free_socket(&connection->data);
2678 	kfree(connection->int_dig_in);
2679 	kfree(connection->int_dig_vv);
2680 	kfree(connection);
2681 	kref_put(&resource->kref, drbd_destroy_resource);
2682 }
2683 
init_submitter(struct drbd_device * device)2684 static int init_submitter(struct drbd_device *device)
2685 {
2686 	/* opencoded create_singlethread_workqueue(),
2687 	 * to be able to say "drbd%d", ..., minor */
2688 	device->submit.wq =
2689 		alloc_ordered_workqueue("drbd%u_submit", WQ_MEM_RECLAIM, device->minor);
2690 	if (!device->submit.wq)
2691 		return -ENOMEM;
2692 
2693 	INIT_WORK(&device->submit.worker, do_submit);
2694 	INIT_LIST_HEAD(&device->submit.writes);
2695 	return 0;
2696 }
2697 
drbd_create_device(struct drbd_config_context * adm_ctx,unsigned int minor)2698 enum drbd_ret_code drbd_create_device(struct drbd_config_context *adm_ctx, unsigned int minor)
2699 {
2700 	struct drbd_resource *resource = adm_ctx->resource;
2701 	struct drbd_connection *connection, *n;
2702 	struct drbd_device *device;
2703 	struct drbd_peer_device *peer_device, *tmp_peer_device;
2704 	struct gendisk *disk;
2705 	int id;
2706 	int vnr = adm_ctx->volume;
2707 	enum drbd_ret_code err = ERR_NOMEM;
2708 
2709 	device = minor_to_device(minor);
2710 	if (device)
2711 		return ERR_MINOR_OR_VOLUME_EXISTS;
2712 
2713 	/* GFP_KERNEL, we are outside of all write-out paths */
2714 	device = kzalloc(sizeof(struct drbd_device), GFP_KERNEL);
2715 	if (!device)
2716 		return ERR_NOMEM;
2717 	kref_init(&device->kref);
2718 
2719 	kref_get(&resource->kref);
2720 	device->resource = resource;
2721 	device->minor = minor;
2722 	device->vnr = vnr;
2723 
2724 	drbd_init_set_defaults(device);
2725 
2726 	disk = blk_alloc_disk(NUMA_NO_NODE);
2727 	if (!disk)
2728 		goto out_no_disk;
2729 
2730 	device->vdisk = disk;
2731 	device->rq_queue = disk->queue;
2732 
2733 	set_disk_ro(disk, true);
2734 
2735 	disk->major = DRBD_MAJOR;
2736 	disk->first_minor = minor;
2737 	disk->minors = 1;
2738 	disk->fops = &drbd_ops;
2739 	sprintf(disk->disk_name, "drbd%d", minor);
2740 	disk->private_data = device;
2741 
2742 	blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, disk->queue);
2743 	blk_queue_write_cache(disk->queue, true, true);
2744 	/* Setting the max_hw_sectors to an odd value of 8kibyte here
2745 	   This triggers a max_bio_size message upon first attach or connect */
2746 	blk_queue_max_hw_sectors(disk->queue, DRBD_MAX_BIO_SIZE_SAFE >> 8);
2747 
2748 	device->md_io.page = alloc_page(GFP_KERNEL);
2749 	if (!device->md_io.page)
2750 		goto out_no_io_page;
2751 
2752 	if (drbd_bm_init(device))
2753 		goto out_no_bitmap;
2754 	device->read_requests = RB_ROOT;
2755 	device->write_requests = RB_ROOT;
2756 
2757 	id = idr_alloc(&drbd_devices, device, minor, minor + 1, GFP_KERNEL);
2758 	if (id < 0) {
2759 		if (id == -ENOSPC)
2760 			err = ERR_MINOR_OR_VOLUME_EXISTS;
2761 		goto out_no_minor_idr;
2762 	}
2763 	kref_get(&device->kref);
2764 
2765 	id = idr_alloc(&resource->devices, device, vnr, vnr + 1, GFP_KERNEL);
2766 	if (id < 0) {
2767 		if (id == -ENOSPC)
2768 			err = ERR_MINOR_OR_VOLUME_EXISTS;
2769 		goto out_idr_remove_minor;
2770 	}
2771 	kref_get(&device->kref);
2772 
2773 	INIT_LIST_HEAD(&device->peer_devices);
2774 	INIT_LIST_HEAD(&device->pending_bitmap_io);
2775 	for_each_connection(connection, resource) {
2776 		peer_device = kzalloc(sizeof(struct drbd_peer_device), GFP_KERNEL);
2777 		if (!peer_device)
2778 			goto out_idr_remove_from_resource;
2779 		peer_device->connection = connection;
2780 		peer_device->device = device;
2781 
2782 		list_add(&peer_device->peer_devices, &device->peer_devices);
2783 		kref_get(&device->kref);
2784 
2785 		id = idr_alloc(&connection->peer_devices, peer_device, vnr, vnr + 1, GFP_KERNEL);
2786 		if (id < 0) {
2787 			if (id == -ENOSPC)
2788 				err = ERR_INVALID_REQUEST;
2789 			goto out_idr_remove_from_resource;
2790 		}
2791 		kref_get(&connection->kref);
2792 		INIT_WORK(&peer_device->send_acks_work, drbd_send_acks_wf);
2793 	}
2794 
2795 	if (init_submitter(device)) {
2796 		err = ERR_NOMEM;
2797 		goto out_idr_remove_from_resource;
2798 	}
2799 
2800 	err = add_disk(disk);
2801 	if (err)
2802 		goto out_destroy_workqueue;
2803 
2804 	/* inherit the connection state */
2805 	device->state.conn = first_connection(resource)->cstate;
2806 	if (device->state.conn == C_WF_REPORT_PARAMS) {
2807 		for_each_peer_device(peer_device, device)
2808 			drbd_connected(peer_device);
2809 	}
2810 	/* move to create_peer_device() */
2811 	for_each_peer_device(peer_device, device)
2812 		drbd_debugfs_peer_device_add(peer_device);
2813 	drbd_debugfs_device_add(device);
2814 	return NO_ERROR;
2815 
2816 out_destroy_workqueue:
2817 	destroy_workqueue(device->submit.wq);
2818 out_idr_remove_from_resource:
2819 	for_each_connection_safe(connection, n, resource) {
2820 		peer_device = idr_remove(&connection->peer_devices, vnr);
2821 		if (peer_device)
2822 			kref_put(&connection->kref, drbd_destroy_connection);
2823 	}
2824 	for_each_peer_device_safe(peer_device, tmp_peer_device, device) {
2825 		list_del(&peer_device->peer_devices);
2826 		kfree(peer_device);
2827 	}
2828 	idr_remove(&resource->devices, vnr);
2829 out_idr_remove_minor:
2830 	idr_remove(&drbd_devices, minor);
2831 	synchronize_rcu();
2832 out_no_minor_idr:
2833 	drbd_bm_cleanup(device);
2834 out_no_bitmap:
2835 	__free_page(device->md_io.page);
2836 out_no_io_page:
2837 	blk_cleanup_disk(disk);
2838 out_no_disk:
2839 	kref_put(&resource->kref, drbd_destroy_resource);
2840 	kfree(device);
2841 	return err;
2842 }
2843 
drbd_delete_device(struct drbd_device * device)2844 void drbd_delete_device(struct drbd_device *device)
2845 {
2846 	struct drbd_resource *resource = device->resource;
2847 	struct drbd_connection *connection;
2848 	struct drbd_peer_device *peer_device;
2849 
2850 	/* move to free_peer_device() */
2851 	for_each_peer_device(peer_device, device)
2852 		drbd_debugfs_peer_device_cleanup(peer_device);
2853 	drbd_debugfs_device_cleanup(device);
2854 	for_each_connection(connection, resource) {
2855 		idr_remove(&connection->peer_devices, device->vnr);
2856 		kref_put(&device->kref, drbd_destroy_device);
2857 	}
2858 	idr_remove(&resource->devices, device->vnr);
2859 	kref_put(&device->kref, drbd_destroy_device);
2860 	idr_remove(&drbd_devices, device_to_minor(device));
2861 	kref_put(&device->kref, drbd_destroy_device);
2862 	del_gendisk(device->vdisk);
2863 	synchronize_rcu();
2864 	kref_put(&device->kref, drbd_destroy_device);
2865 }
2866 
drbd_init(void)2867 static int __init drbd_init(void)
2868 {
2869 	int err;
2870 
2871 	if (drbd_minor_count < DRBD_MINOR_COUNT_MIN || drbd_minor_count > DRBD_MINOR_COUNT_MAX) {
2872 		pr_err("invalid minor_count (%d)\n", drbd_minor_count);
2873 #ifdef MODULE
2874 		return -EINVAL;
2875 #else
2876 		drbd_minor_count = DRBD_MINOR_COUNT_DEF;
2877 #endif
2878 	}
2879 
2880 	err = register_blkdev(DRBD_MAJOR, "drbd");
2881 	if (err) {
2882 		pr_err("unable to register block device major %d\n",
2883 		       DRBD_MAJOR);
2884 		return err;
2885 	}
2886 
2887 	/*
2888 	 * allocate all necessary structs
2889 	 */
2890 	init_waitqueue_head(&drbd_pp_wait);
2891 
2892 	drbd_proc = NULL; /* play safe for drbd_cleanup */
2893 	idr_init(&drbd_devices);
2894 
2895 	mutex_init(&resources_mutex);
2896 	INIT_LIST_HEAD(&drbd_resources);
2897 
2898 	err = drbd_genl_register();
2899 	if (err) {
2900 		pr_err("unable to register generic netlink family\n");
2901 		goto fail;
2902 	}
2903 
2904 	err = drbd_create_mempools();
2905 	if (err)
2906 		goto fail;
2907 
2908 	err = -ENOMEM;
2909 	drbd_proc = proc_create_single("drbd", S_IFREG | 0444 , NULL, drbd_seq_show);
2910 	if (!drbd_proc)	{
2911 		pr_err("unable to register proc file\n");
2912 		goto fail;
2913 	}
2914 
2915 	retry.wq = create_singlethread_workqueue("drbd-reissue");
2916 	if (!retry.wq) {
2917 		pr_err("unable to create retry workqueue\n");
2918 		goto fail;
2919 	}
2920 	INIT_WORK(&retry.worker, do_retry);
2921 	spin_lock_init(&retry.lock);
2922 	INIT_LIST_HEAD(&retry.writes);
2923 
2924 	drbd_debugfs_init();
2925 
2926 	pr_info("initialized. "
2927 	       "Version: " REL_VERSION " (api:%d/proto:%d-%d)\n",
2928 	       API_VERSION, PRO_VERSION_MIN, PRO_VERSION_MAX);
2929 	pr_info("%s\n", drbd_buildtag());
2930 	pr_info("registered as block device major %d\n", DRBD_MAJOR);
2931 	return 0; /* Success! */
2932 
2933 fail:
2934 	drbd_cleanup();
2935 	if (err == -ENOMEM)
2936 		pr_err("ran out of memory\n");
2937 	else
2938 		pr_err("initialization failure\n");
2939 	return err;
2940 }
2941 
drbd_free_one_sock(struct drbd_socket * ds)2942 static void drbd_free_one_sock(struct drbd_socket *ds)
2943 {
2944 	struct socket *s;
2945 	mutex_lock(&ds->mutex);
2946 	s = ds->socket;
2947 	ds->socket = NULL;
2948 	mutex_unlock(&ds->mutex);
2949 	if (s) {
2950 		/* so debugfs does not need to mutex_lock() */
2951 		synchronize_rcu();
2952 		kernel_sock_shutdown(s, SHUT_RDWR);
2953 		sock_release(s);
2954 	}
2955 }
2956 
drbd_free_sock(struct drbd_connection * connection)2957 void drbd_free_sock(struct drbd_connection *connection)
2958 {
2959 	if (connection->data.socket)
2960 		drbd_free_one_sock(&connection->data);
2961 	if (connection->meta.socket)
2962 		drbd_free_one_sock(&connection->meta);
2963 }
2964 
2965 /* meta data management */
2966 
conn_md_sync(struct drbd_connection * connection)2967 void conn_md_sync(struct drbd_connection *connection)
2968 {
2969 	struct drbd_peer_device *peer_device;
2970 	int vnr;
2971 
2972 	rcu_read_lock();
2973 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
2974 		struct drbd_device *device = peer_device->device;
2975 
2976 		kref_get(&device->kref);
2977 		rcu_read_unlock();
2978 		drbd_md_sync(device);
2979 		kref_put(&device->kref, drbd_destroy_device);
2980 		rcu_read_lock();
2981 	}
2982 	rcu_read_unlock();
2983 }
2984 
2985 /* aligned 4kByte */
2986 struct meta_data_on_disk {
2987 	u64 la_size_sect;      /* last agreed size. */
2988 	u64 uuid[UI_SIZE];   /* UUIDs. */
2989 	u64 device_uuid;
2990 	u64 reserved_u64_1;
2991 	u32 flags;             /* MDF */
2992 	u32 magic;
2993 	u32 md_size_sect;
2994 	u32 al_offset;         /* offset to this block */
2995 	u32 al_nr_extents;     /* important for restoring the AL (userspace) */
2996 	      /* `-- act_log->nr_elements <-- ldev->dc.al_extents */
2997 	u32 bm_offset;         /* offset to the bitmap, from here */
2998 	u32 bm_bytes_per_bit;  /* BM_BLOCK_SIZE */
2999 	u32 la_peer_max_bio_size;   /* last peer max_bio_size */
3000 
3001 	/* see al_tr_number_to_on_disk_sector() */
3002 	u32 al_stripes;
3003 	u32 al_stripe_size_4k;
3004 
3005 	u8 reserved_u8[4096 - (7*8 + 10*4)];
3006 } __packed;
3007 
3008 
3009 
drbd_md_write(struct drbd_device * device,void * b)3010 void drbd_md_write(struct drbd_device *device, void *b)
3011 {
3012 	struct meta_data_on_disk *buffer = b;
3013 	sector_t sector;
3014 	int i;
3015 
3016 	memset(buffer, 0, sizeof(*buffer));
3017 
3018 	buffer->la_size_sect = cpu_to_be64(get_capacity(device->vdisk));
3019 	for (i = UI_CURRENT; i < UI_SIZE; i++)
3020 		buffer->uuid[i] = cpu_to_be64(device->ldev->md.uuid[i]);
3021 	buffer->flags = cpu_to_be32(device->ldev->md.flags);
3022 	buffer->magic = cpu_to_be32(DRBD_MD_MAGIC_84_UNCLEAN);
3023 
3024 	buffer->md_size_sect  = cpu_to_be32(device->ldev->md.md_size_sect);
3025 	buffer->al_offset     = cpu_to_be32(device->ldev->md.al_offset);
3026 	buffer->al_nr_extents = cpu_to_be32(device->act_log->nr_elements);
3027 	buffer->bm_bytes_per_bit = cpu_to_be32(BM_BLOCK_SIZE);
3028 	buffer->device_uuid = cpu_to_be64(device->ldev->md.device_uuid);
3029 
3030 	buffer->bm_offset = cpu_to_be32(device->ldev->md.bm_offset);
3031 	buffer->la_peer_max_bio_size = cpu_to_be32(device->peer_max_bio_size);
3032 
3033 	buffer->al_stripes = cpu_to_be32(device->ldev->md.al_stripes);
3034 	buffer->al_stripe_size_4k = cpu_to_be32(device->ldev->md.al_stripe_size_4k);
3035 
3036 	D_ASSERT(device, drbd_md_ss(device->ldev) == device->ldev->md.md_offset);
3037 	sector = device->ldev->md.md_offset;
3038 
3039 	if (drbd_md_sync_page_io(device, device->ldev, sector, REQ_OP_WRITE)) {
3040 		/* this was a try anyways ... */
3041 		drbd_err(device, "meta data update failed!\n");
3042 		drbd_chk_io_error(device, 1, DRBD_META_IO_ERROR);
3043 	}
3044 }
3045 
3046 /**
3047  * drbd_md_sync() - Writes the meta data super block if the MD_DIRTY flag bit is set
3048  * @device:	DRBD device.
3049  */
drbd_md_sync(struct drbd_device * device)3050 void drbd_md_sync(struct drbd_device *device)
3051 {
3052 	struct meta_data_on_disk *buffer;
3053 
3054 	/* Don't accidentally change the DRBD meta data layout. */
3055 	BUILD_BUG_ON(UI_SIZE != 4);
3056 	BUILD_BUG_ON(sizeof(struct meta_data_on_disk) != 4096);
3057 
3058 	del_timer(&device->md_sync_timer);
3059 	/* timer may be rearmed by drbd_md_mark_dirty() now. */
3060 	if (!test_and_clear_bit(MD_DIRTY, &device->flags))
3061 		return;
3062 
3063 	/* We use here D_FAILED and not D_ATTACHING because we try to write
3064 	 * metadata even if we detach due to a disk failure! */
3065 	if (!get_ldev_if_state(device, D_FAILED))
3066 		return;
3067 
3068 	buffer = drbd_md_get_buffer(device, __func__);
3069 	if (!buffer)
3070 		goto out;
3071 
3072 	drbd_md_write(device, buffer);
3073 
3074 	/* Update device->ldev->md.la_size_sect,
3075 	 * since we updated it on metadata. */
3076 	device->ldev->md.la_size_sect = get_capacity(device->vdisk);
3077 
3078 	drbd_md_put_buffer(device);
3079 out:
3080 	put_ldev(device);
3081 }
3082 
check_activity_log_stripe_size(struct drbd_device * device,struct meta_data_on_disk * on_disk,struct drbd_md * in_core)3083 static int check_activity_log_stripe_size(struct drbd_device *device,
3084 		struct meta_data_on_disk *on_disk,
3085 		struct drbd_md *in_core)
3086 {
3087 	u32 al_stripes = be32_to_cpu(on_disk->al_stripes);
3088 	u32 al_stripe_size_4k = be32_to_cpu(on_disk->al_stripe_size_4k);
3089 	u64 al_size_4k;
3090 
3091 	/* both not set: default to old fixed size activity log */
3092 	if (al_stripes == 0 && al_stripe_size_4k == 0) {
3093 		al_stripes = 1;
3094 		al_stripe_size_4k = MD_32kB_SECT/8;
3095 	}
3096 
3097 	/* some paranoia plausibility checks */
3098 
3099 	/* we need both values to be set */
3100 	if (al_stripes == 0 || al_stripe_size_4k == 0)
3101 		goto err;
3102 
3103 	al_size_4k = (u64)al_stripes * al_stripe_size_4k;
3104 
3105 	/* Upper limit of activity log area, to avoid potential overflow
3106 	 * problems in al_tr_number_to_on_disk_sector(). As right now, more
3107 	 * than 72 * 4k blocks total only increases the amount of history,
3108 	 * limiting this arbitrarily to 16 GB is not a real limitation ;-)  */
3109 	if (al_size_4k > (16 * 1024 * 1024/4))
3110 		goto err;
3111 
3112 	/* Lower limit: we need at least 8 transaction slots (32kB)
3113 	 * to not break existing setups */
3114 	if (al_size_4k < MD_32kB_SECT/8)
3115 		goto err;
3116 
3117 	in_core->al_stripe_size_4k = al_stripe_size_4k;
3118 	in_core->al_stripes = al_stripes;
3119 	in_core->al_size_4k = al_size_4k;
3120 
3121 	return 0;
3122 err:
3123 	drbd_err(device, "invalid activity log striping: al_stripes=%u, al_stripe_size_4k=%u\n",
3124 			al_stripes, al_stripe_size_4k);
3125 	return -EINVAL;
3126 }
3127 
check_offsets_and_sizes(struct drbd_device * device,struct drbd_backing_dev * bdev)3128 static int check_offsets_and_sizes(struct drbd_device *device, struct drbd_backing_dev *bdev)
3129 {
3130 	sector_t capacity = drbd_get_capacity(bdev->md_bdev);
3131 	struct drbd_md *in_core = &bdev->md;
3132 	s32 on_disk_al_sect;
3133 	s32 on_disk_bm_sect;
3134 
3135 	/* The on-disk size of the activity log, calculated from offsets, and
3136 	 * the size of the activity log calculated from the stripe settings,
3137 	 * should match.
3138 	 * Though we could relax this a bit: it is ok, if the striped activity log
3139 	 * fits in the available on-disk activity log size.
3140 	 * Right now, that would break how resize is implemented.
3141 	 * TODO: make drbd_determine_dev_size() (and the drbdmeta tool) aware
3142 	 * of possible unused padding space in the on disk layout. */
3143 	if (in_core->al_offset < 0) {
3144 		if (in_core->bm_offset > in_core->al_offset)
3145 			goto err;
3146 		on_disk_al_sect = -in_core->al_offset;
3147 		on_disk_bm_sect = in_core->al_offset - in_core->bm_offset;
3148 	} else {
3149 		if (in_core->al_offset != MD_4kB_SECT)
3150 			goto err;
3151 		if (in_core->bm_offset < in_core->al_offset + in_core->al_size_4k * MD_4kB_SECT)
3152 			goto err;
3153 
3154 		on_disk_al_sect = in_core->bm_offset - MD_4kB_SECT;
3155 		on_disk_bm_sect = in_core->md_size_sect - in_core->bm_offset;
3156 	}
3157 
3158 	/* old fixed size meta data is exactly that: fixed. */
3159 	if (in_core->meta_dev_idx >= 0) {
3160 		if (in_core->md_size_sect != MD_128MB_SECT
3161 		||  in_core->al_offset != MD_4kB_SECT
3162 		||  in_core->bm_offset != MD_4kB_SECT + MD_32kB_SECT
3163 		||  in_core->al_stripes != 1
3164 		||  in_core->al_stripe_size_4k != MD_32kB_SECT/8)
3165 			goto err;
3166 	}
3167 
3168 	if (capacity < in_core->md_size_sect)
3169 		goto err;
3170 	if (capacity - in_core->md_size_sect < drbd_md_first_sector(bdev))
3171 		goto err;
3172 
3173 	/* should be aligned, and at least 32k */
3174 	if ((on_disk_al_sect & 7) || (on_disk_al_sect < MD_32kB_SECT))
3175 		goto err;
3176 
3177 	/* should fit (for now: exactly) into the available on-disk space;
3178 	 * overflow prevention is in check_activity_log_stripe_size() above. */
3179 	if (on_disk_al_sect != in_core->al_size_4k * MD_4kB_SECT)
3180 		goto err;
3181 
3182 	/* again, should be aligned */
3183 	if (in_core->bm_offset & 7)
3184 		goto err;
3185 
3186 	/* FIXME check for device grow with flex external meta data? */
3187 
3188 	/* can the available bitmap space cover the last agreed device size? */
3189 	if (on_disk_bm_sect < (in_core->la_size_sect+7)/MD_4kB_SECT/8/512)
3190 		goto err;
3191 
3192 	return 0;
3193 
3194 err:
3195 	drbd_err(device, "meta data offsets don't make sense: idx=%d "
3196 			"al_s=%u, al_sz4k=%u, al_offset=%d, bm_offset=%d, "
3197 			"md_size_sect=%u, la_size=%llu, md_capacity=%llu\n",
3198 			in_core->meta_dev_idx,
3199 			in_core->al_stripes, in_core->al_stripe_size_4k,
3200 			in_core->al_offset, in_core->bm_offset, in_core->md_size_sect,
3201 			(unsigned long long)in_core->la_size_sect,
3202 			(unsigned long long)capacity);
3203 
3204 	return -EINVAL;
3205 }
3206 
3207 
3208 /**
3209  * drbd_md_read() - Reads in the meta data super block
3210  * @device:	DRBD device.
3211  * @bdev:	Device from which the meta data should be read in.
3212  *
3213  * Return NO_ERROR on success, and an enum drbd_ret_code in case
3214  * something goes wrong.
3215  *
3216  * Called exactly once during drbd_adm_attach(), while still being D_DISKLESS,
3217  * even before @bdev is assigned to @device->ldev.
3218  */
drbd_md_read(struct drbd_device * device,struct drbd_backing_dev * bdev)3219 int drbd_md_read(struct drbd_device *device, struct drbd_backing_dev *bdev)
3220 {
3221 	struct meta_data_on_disk *buffer;
3222 	u32 magic, flags;
3223 	int i, rv = NO_ERROR;
3224 
3225 	if (device->state.disk != D_DISKLESS)
3226 		return ERR_DISK_CONFIGURED;
3227 
3228 	buffer = drbd_md_get_buffer(device, __func__);
3229 	if (!buffer)
3230 		return ERR_NOMEM;
3231 
3232 	/* First, figure out where our meta data superblock is located,
3233 	 * and read it. */
3234 	bdev->md.meta_dev_idx = bdev->disk_conf->meta_dev_idx;
3235 	bdev->md.md_offset = drbd_md_ss(bdev);
3236 	/* Even for (flexible or indexed) external meta data,
3237 	 * initially restrict us to the 4k superblock for now.
3238 	 * Affects the paranoia out-of-range access check in drbd_md_sync_page_io(). */
3239 	bdev->md.md_size_sect = 8;
3240 
3241 	if (drbd_md_sync_page_io(device, bdev, bdev->md.md_offset,
3242 				 REQ_OP_READ)) {
3243 		/* NOTE: can't do normal error processing here as this is
3244 		   called BEFORE disk is attached */
3245 		drbd_err(device, "Error while reading metadata.\n");
3246 		rv = ERR_IO_MD_DISK;
3247 		goto err;
3248 	}
3249 
3250 	magic = be32_to_cpu(buffer->magic);
3251 	flags = be32_to_cpu(buffer->flags);
3252 	if (magic == DRBD_MD_MAGIC_84_UNCLEAN ||
3253 	    (magic == DRBD_MD_MAGIC_08 && !(flags & MDF_AL_CLEAN))) {
3254 			/* btw: that's Activity Log clean, not "all" clean. */
3255 		drbd_err(device, "Found unclean meta data. Did you \"drbdadm apply-al\"?\n");
3256 		rv = ERR_MD_UNCLEAN;
3257 		goto err;
3258 	}
3259 
3260 	rv = ERR_MD_INVALID;
3261 	if (magic != DRBD_MD_MAGIC_08) {
3262 		if (magic == DRBD_MD_MAGIC_07)
3263 			drbd_err(device, "Found old (0.7) meta data magic. Did you \"drbdadm create-md\"?\n");
3264 		else
3265 			drbd_err(device, "Meta data magic not found. Did you \"drbdadm create-md\"?\n");
3266 		goto err;
3267 	}
3268 
3269 	if (be32_to_cpu(buffer->bm_bytes_per_bit) != BM_BLOCK_SIZE) {
3270 		drbd_err(device, "unexpected bm_bytes_per_bit: %u (expected %u)\n",
3271 		    be32_to_cpu(buffer->bm_bytes_per_bit), BM_BLOCK_SIZE);
3272 		goto err;
3273 	}
3274 
3275 
3276 	/* convert to in_core endian */
3277 	bdev->md.la_size_sect = be64_to_cpu(buffer->la_size_sect);
3278 	for (i = UI_CURRENT; i < UI_SIZE; i++)
3279 		bdev->md.uuid[i] = be64_to_cpu(buffer->uuid[i]);
3280 	bdev->md.flags = be32_to_cpu(buffer->flags);
3281 	bdev->md.device_uuid = be64_to_cpu(buffer->device_uuid);
3282 
3283 	bdev->md.md_size_sect = be32_to_cpu(buffer->md_size_sect);
3284 	bdev->md.al_offset = be32_to_cpu(buffer->al_offset);
3285 	bdev->md.bm_offset = be32_to_cpu(buffer->bm_offset);
3286 
3287 	if (check_activity_log_stripe_size(device, buffer, &bdev->md))
3288 		goto err;
3289 	if (check_offsets_and_sizes(device, bdev))
3290 		goto err;
3291 
3292 	if (be32_to_cpu(buffer->bm_offset) != bdev->md.bm_offset) {
3293 		drbd_err(device, "unexpected bm_offset: %d (expected %d)\n",
3294 		    be32_to_cpu(buffer->bm_offset), bdev->md.bm_offset);
3295 		goto err;
3296 	}
3297 	if (be32_to_cpu(buffer->md_size_sect) != bdev->md.md_size_sect) {
3298 		drbd_err(device, "unexpected md_size: %u (expected %u)\n",
3299 		    be32_to_cpu(buffer->md_size_sect), bdev->md.md_size_sect);
3300 		goto err;
3301 	}
3302 
3303 	rv = NO_ERROR;
3304 
3305 	spin_lock_irq(&device->resource->req_lock);
3306 	if (device->state.conn < C_CONNECTED) {
3307 		unsigned int peer;
3308 		peer = be32_to_cpu(buffer->la_peer_max_bio_size);
3309 		peer = max(peer, DRBD_MAX_BIO_SIZE_SAFE);
3310 		device->peer_max_bio_size = peer;
3311 	}
3312 	spin_unlock_irq(&device->resource->req_lock);
3313 
3314  err:
3315 	drbd_md_put_buffer(device);
3316 
3317 	return rv;
3318 }
3319 
3320 /**
3321  * drbd_md_mark_dirty() - Mark meta data super block as dirty
3322  * @device:	DRBD device.
3323  *
3324  * Call this function if you change anything that should be written to
3325  * the meta-data super block. This function sets MD_DIRTY, and starts a
3326  * timer that ensures that within five seconds you have to call drbd_md_sync().
3327  */
drbd_md_mark_dirty(struct drbd_device * device)3328 void drbd_md_mark_dirty(struct drbd_device *device)
3329 {
3330 	if (!test_and_set_bit(MD_DIRTY, &device->flags))
3331 		mod_timer(&device->md_sync_timer, jiffies + 5*HZ);
3332 }
3333 
drbd_uuid_move_history(struct drbd_device * device)3334 void drbd_uuid_move_history(struct drbd_device *device) __must_hold(local)
3335 {
3336 	int i;
3337 
3338 	for (i = UI_HISTORY_START; i < UI_HISTORY_END; i++)
3339 		device->ldev->md.uuid[i+1] = device->ldev->md.uuid[i];
3340 }
3341 
__drbd_uuid_set(struct drbd_device * device,int idx,u64 val)3342 void __drbd_uuid_set(struct drbd_device *device, int idx, u64 val) __must_hold(local)
3343 {
3344 	if (idx == UI_CURRENT) {
3345 		if (device->state.role == R_PRIMARY)
3346 			val |= 1;
3347 		else
3348 			val &= ~((u64)1);
3349 
3350 		drbd_set_ed_uuid(device, val);
3351 	}
3352 
3353 	device->ldev->md.uuid[idx] = val;
3354 	drbd_md_mark_dirty(device);
3355 }
3356 
_drbd_uuid_set(struct drbd_device * device,int idx,u64 val)3357 void _drbd_uuid_set(struct drbd_device *device, int idx, u64 val) __must_hold(local)
3358 {
3359 	unsigned long flags;
3360 	spin_lock_irqsave(&device->ldev->md.uuid_lock, flags);
3361 	__drbd_uuid_set(device, idx, val);
3362 	spin_unlock_irqrestore(&device->ldev->md.uuid_lock, flags);
3363 }
3364 
drbd_uuid_set(struct drbd_device * device,int idx,u64 val)3365 void drbd_uuid_set(struct drbd_device *device, int idx, u64 val) __must_hold(local)
3366 {
3367 	unsigned long flags;
3368 	spin_lock_irqsave(&device->ldev->md.uuid_lock, flags);
3369 	if (device->ldev->md.uuid[idx]) {
3370 		drbd_uuid_move_history(device);
3371 		device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[idx];
3372 	}
3373 	__drbd_uuid_set(device, idx, val);
3374 	spin_unlock_irqrestore(&device->ldev->md.uuid_lock, flags);
3375 }
3376 
3377 /**
3378  * drbd_uuid_new_current() - Creates a new current UUID
3379  * @device:	DRBD device.
3380  *
3381  * Creates a new current UUID, and rotates the old current UUID into
3382  * the bitmap slot. Causes an incremental resync upon next connect.
3383  */
drbd_uuid_new_current(struct drbd_device * device)3384 void drbd_uuid_new_current(struct drbd_device *device) __must_hold(local)
3385 {
3386 	u64 val;
3387 	unsigned long long bm_uuid;
3388 
3389 	get_random_bytes(&val, sizeof(u64));
3390 
3391 	spin_lock_irq(&device->ldev->md.uuid_lock);
3392 	bm_uuid = device->ldev->md.uuid[UI_BITMAP];
3393 
3394 	if (bm_uuid)
3395 		drbd_warn(device, "bm UUID was already set: %llX\n", bm_uuid);
3396 
3397 	device->ldev->md.uuid[UI_BITMAP] = device->ldev->md.uuid[UI_CURRENT];
3398 	__drbd_uuid_set(device, UI_CURRENT, val);
3399 	spin_unlock_irq(&device->ldev->md.uuid_lock);
3400 
3401 	drbd_print_uuids(device, "new current UUID");
3402 	/* get it to stable storage _now_ */
3403 	drbd_md_sync(device);
3404 }
3405 
drbd_uuid_set_bm(struct drbd_device * device,u64 val)3406 void drbd_uuid_set_bm(struct drbd_device *device, u64 val) __must_hold(local)
3407 {
3408 	unsigned long flags;
3409 	if (device->ldev->md.uuid[UI_BITMAP] == 0 && val == 0)
3410 		return;
3411 
3412 	spin_lock_irqsave(&device->ldev->md.uuid_lock, flags);
3413 	if (val == 0) {
3414 		drbd_uuid_move_history(device);
3415 		device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3416 		device->ldev->md.uuid[UI_BITMAP] = 0;
3417 	} else {
3418 		unsigned long long bm_uuid = device->ldev->md.uuid[UI_BITMAP];
3419 		if (bm_uuid)
3420 			drbd_warn(device, "bm UUID was already set: %llX\n", bm_uuid);
3421 
3422 		device->ldev->md.uuid[UI_BITMAP] = val & ~((u64)1);
3423 	}
3424 	spin_unlock_irqrestore(&device->ldev->md.uuid_lock, flags);
3425 
3426 	drbd_md_mark_dirty(device);
3427 }
3428 
3429 /**
3430  * drbd_bmio_set_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
3431  * @device:	DRBD device.
3432  *
3433  * Sets all bits in the bitmap and writes the whole bitmap to stable storage.
3434  */
drbd_bmio_set_n_write(struct drbd_device * device)3435 int drbd_bmio_set_n_write(struct drbd_device *device) __must_hold(local)
3436 {
3437 	int rv = -EIO;
3438 
3439 	drbd_md_set_flag(device, MDF_FULL_SYNC);
3440 	drbd_md_sync(device);
3441 	drbd_bm_set_all(device);
3442 
3443 	rv = drbd_bm_write(device);
3444 
3445 	if (!rv) {
3446 		drbd_md_clear_flag(device, MDF_FULL_SYNC);
3447 		drbd_md_sync(device);
3448 	}
3449 
3450 	return rv;
3451 }
3452 
3453 /**
3454  * drbd_bmio_clear_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
3455  * @device:	DRBD device.
3456  *
3457  * Clears all bits in the bitmap and writes the whole bitmap to stable storage.
3458  */
drbd_bmio_clear_n_write(struct drbd_device * device)3459 int drbd_bmio_clear_n_write(struct drbd_device *device) __must_hold(local)
3460 {
3461 	drbd_resume_al(device);
3462 	drbd_bm_clear_all(device);
3463 	return drbd_bm_write(device);
3464 }
3465 
w_bitmap_io(struct drbd_work * w,int unused)3466 static int w_bitmap_io(struct drbd_work *w, int unused)
3467 {
3468 	struct drbd_device *device =
3469 		container_of(w, struct drbd_device, bm_io_work.w);
3470 	struct bm_io_work *work = &device->bm_io_work;
3471 	int rv = -EIO;
3472 
3473 	if (work->flags != BM_LOCKED_CHANGE_ALLOWED) {
3474 		int cnt = atomic_read(&device->ap_bio_cnt);
3475 		if (cnt)
3476 			drbd_err(device, "FIXME: ap_bio_cnt %d, expected 0; queued for '%s'\n",
3477 					cnt, work->why);
3478 	}
3479 
3480 	if (get_ldev(device)) {
3481 		drbd_bm_lock(device, work->why, work->flags);
3482 		rv = work->io_fn(device);
3483 		drbd_bm_unlock(device);
3484 		put_ldev(device);
3485 	}
3486 
3487 	clear_bit_unlock(BITMAP_IO, &device->flags);
3488 	wake_up(&device->misc_wait);
3489 
3490 	if (work->done)
3491 		work->done(device, rv);
3492 
3493 	clear_bit(BITMAP_IO_QUEUED, &device->flags);
3494 	work->why = NULL;
3495 	work->flags = 0;
3496 
3497 	return 0;
3498 }
3499 
3500 /**
3501  * drbd_queue_bitmap_io() - Queues an IO operation on the whole bitmap
3502  * @device:	DRBD device.
3503  * @io_fn:	IO callback to be called when bitmap IO is possible
3504  * @done:	callback to be called after the bitmap IO was performed
3505  * @why:	Descriptive text of the reason for doing the IO
3506  * @flags:	Bitmap flags
3507  *
3508  * While IO on the bitmap happens we freeze application IO thus we ensure
3509  * that drbd_set_out_of_sync() can not be called. This function MAY ONLY be
3510  * called from worker context. It MUST NOT be used while a previous such
3511  * work is still pending!
3512  *
3513  * Its worker function encloses the call of io_fn() by get_ldev() and
3514  * put_ldev().
3515  */
drbd_queue_bitmap_io(struct drbd_device * device,int (* io_fn)(struct drbd_device *),void (* done)(struct drbd_device *,int),char * why,enum bm_flag flags)3516 void drbd_queue_bitmap_io(struct drbd_device *device,
3517 			  int (*io_fn)(struct drbd_device *),
3518 			  void (*done)(struct drbd_device *, int),
3519 			  char *why, enum bm_flag flags)
3520 {
3521 	D_ASSERT(device, current == first_peer_device(device)->connection->worker.task);
3522 
3523 	D_ASSERT(device, !test_bit(BITMAP_IO_QUEUED, &device->flags));
3524 	D_ASSERT(device, !test_bit(BITMAP_IO, &device->flags));
3525 	D_ASSERT(device, list_empty(&device->bm_io_work.w.list));
3526 	if (device->bm_io_work.why)
3527 		drbd_err(device, "FIXME going to queue '%s' but '%s' still pending?\n",
3528 			why, device->bm_io_work.why);
3529 
3530 	device->bm_io_work.io_fn = io_fn;
3531 	device->bm_io_work.done = done;
3532 	device->bm_io_work.why = why;
3533 	device->bm_io_work.flags = flags;
3534 
3535 	spin_lock_irq(&device->resource->req_lock);
3536 	set_bit(BITMAP_IO, &device->flags);
3537 	/* don't wait for pending application IO if the caller indicates that
3538 	 * application IO does not conflict anyways. */
3539 	if (flags == BM_LOCKED_CHANGE_ALLOWED || atomic_read(&device->ap_bio_cnt) == 0) {
3540 		if (!test_and_set_bit(BITMAP_IO_QUEUED, &device->flags))
3541 			drbd_queue_work(&first_peer_device(device)->connection->sender_work,
3542 					&device->bm_io_work.w);
3543 	}
3544 	spin_unlock_irq(&device->resource->req_lock);
3545 }
3546 
3547 /**
3548  * drbd_bitmap_io() -  Does an IO operation on the whole bitmap
3549  * @device:	DRBD device.
3550  * @io_fn:	IO callback to be called when bitmap IO is possible
3551  * @why:	Descriptive text of the reason for doing the IO
3552  * @flags:	Bitmap flags
3553  *
3554  * freezes application IO while that the actual IO operations runs. This
3555  * functions MAY NOT be called from worker context.
3556  */
drbd_bitmap_io(struct drbd_device * device,int (* io_fn)(struct drbd_device *),char * why,enum bm_flag flags)3557 int drbd_bitmap_io(struct drbd_device *device, int (*io_fn)(struct drbd_device *),
3558 		char *why, enum bm_flag flags)
3559 {
3560 	/* Only suspend io, if some operation is supposed to be locked out */
3561 	const bool do_suspend_io = flags & (BM_DONT_CLEAR|BM_DONT_SET|BM_DONT_TEST);
3562 	int rv;
3563 
3564 	D_ASSERT(device, current != first_peer_device(device)->connection->worker.task);
3565 
3566 	if (do_suspend_io)
3567 		drbd_suspend_io(device);
3568 
3569 	drbd_bm_lock(device, why, flags);
3570 	rv = io_fn(device);
3571 	drbd_bm_unlock(device);
3572 
3573 	if (do_suspend_io)
3574 		drbd_resume_io(device);
3575 
3576 	return rv;
3577 }
3578 
drbd_md_set_flag(struct drbd_device * device,int flag)3579 void drbd_md_set_flag(struct drbd_device *device, int flag) __must_hold(local)
3580 {
3581 	if ((device->ldev->md.flags & flag) != flag) {
3582 		drbd_md_mark_dirty(device);
3583 		device->ldev->md.flags |= flag;
3584 	}
3585 }
3586 
drbd_md_clear_flag(struct drbd_device * device,int flag)3587 void drbd_md_clear_flag(struct drbd_device *device, int flag) __must_hold(local)
3588 {
3589 	if ((device->ldev->md.flags & flag) != 0) {
3590 		drbd_md_mark_dirty(device);
3591 		device->ldev->md.flags &= ~flag;
3592 	}
3593 }
drbd_md_test_flag(struct drbd_backing_dev * bdev,int flag)3594 int drbd_md_test_flag(struct drbd_backing_dev *bdev, int flag)
3595 {
3596 	return (bdev->md.flags & flag) != 0;
3597 }
3598 
md_sync_timer_fn(struct timer_list * t)3599 static void md_sync_timer_fn(struct timer_list *t)
3600 {
3601 	struct drbd_device *device = from_timer(device, t, md_sync_timer);
3602 	drbd_device_post_work(device, MD_SYNC);
3603 }
3604 
cmdname(enum drbd_packet cmd)3605 const char *cmdname(enum drbd_packet cmd)
3606 {
3607 	/* THINK may need to become several global tables
3608 	 * when we want to support more than
3609 	 * one PRO_VERSION */
3610 	static const char *cmdnames[] = {
3611 
3612 		[P_DATA]	        = "Data",
3613 		[P_DATA_REPLY]	        = "DataReply",
3614 		[P_RS_DATA_REPLY]	= "RSDataReply",
3615 		[P_BARRIER]	        = "Barrier",
3616 		[P_BITMAP]	        = "ReportBitMap",
3617 		[P_BECOME_SYNC_TARGET]  = "BecomeSyncTarget",
3618 		[P_BECOME_SYNC_SOURCE]  = "BecomeSyncSource",
3619 		[P_UNPLUG_REMOTE]	= "UnplugRemote",
3620 		[P_DATA_REQUEST]	= "DataRequest",
3621 		[P_RS_DATA_REQUEST]     = "RSDataRequest",
3622 		[P_SYNC_PARAM]	        = "SyncParam",
3623 		[P_PROTOCOL]            = "ReportProtocol",
3624 		[P_UUIDS]	        = "ReportUUIDs",
3625 		[P_SIZES]	        = "ReportSizes",
3626 		[P_STATE]	        = "ReportState",
3627 		[P_SYNC_UUID]           = "ReportSyncUUID",
3628 		[P_AUTH_CHALLENGE]      = "AuthChallenge",
3629 		[P_AUTH_RESPONSE]	= "AuthResponse",
3630 		[P_STATE_CHG_REQ]       = "StateChgRequest",
3631 		[P_PING]		= "Ping",
3632 		[P_PING_ACK]	        = "PingAck",
3633 		[P_RECV_ACK]	        = "RecvAck",
3634 		[P_WRITE_ACK]	        = "WriteAck",
3635 		[P_RS_WRITE_ACK]	= "RSWriteAck",
3636 		[P_SUPERSEDED]          = "Superseded",
3637 		[P_NEG_ACK]	        = "NegAck",
3638 		[P_NEG_DREPLY]	        = "NegDReply",
3639 		[P_NEG_RS_DREPLY]	= "NegRSDReply",
3640 		[P_BARRIER_ACK]	        = "BarrierAck",
3641 		[P_STATE_CHG_REPLY]     = "StateChgReply",
3642 		[P_OV_REQUEST]          = "OVRequest",
3643 		[P_OV_REPLY]            = "OVReply",
3644 		[P_OV_RESULT]           = "OVResult",
3645 		[P_CSUM_RS_REQUEST]     = "CsumRSRequest",
3646 		[P_RS_IS_IN_SYNC]	= "CsumRSIsInSync",
3647 		[P_SYNC_PARAM89]	= "SyncParam89",
3648 		[P_COMPRESSED_BITMAP]   = "CBitmap",
3649 		[P_DELAY_PROBE]         = "DelayProbe",
3650 		[P_OUT_OF_SYNC]		= "OutOfSync",
3651 		[P_RS_CANCEL]		= "RSCancel",
3652 		[P_CONN_ST_CHG_REQ]	= "conn_st_chg_req",
3653 		[P_CONN_ST_CHG_REPLY]	= "conn_st_chg_reply",
3654 		[P_PROTOCOL_UPDATE]	= "protocol_update",
3655 		[P_TRIM]	        = "Trim",
3656 		[P_RS_THIN_REQ]         = "rs_thin_req",
3657 		[P_RS_DEALLOCATED]      = "rs_deallocated",
3658 		[P_WSAME]	        = "WriteSame",
3659 		[P_ZEROES]		= "Zeroes",
3660 
3661 		/* enum drbd_packet, but not commands - obsoleted flags:
3662 		 *	P_MAY_IGNORE
3663 		 *	P_MAX_OPT_CMD
3664 		 */
3665 	};
3666 
3667 	/* too big for the array: 0xfffX */
3668 	if (cmd == P_INITIAL_META)
3669 		return "InitialMeta";
3670 	if (cmd == P_INITIAL_DATA)
3671 		return "InitialData";
3672 	if (cmd == P_CONNECTION_FEATURES)
3673 		return "ConnectionFeatures";
3674 	if (cmd >= ARRAY_SIZE(cmdnames))
3675 		return "Unknown";
3676 	return cmdnames[cmd];
3677 }
3678 
3679 /**
3680  * drbd_wait_misc  -  wait for a request to make progress
3681  * @device:	device associated with the request
3682  * @i:		the struct drbd_interval embedded in struct drbd_request or
3683  *		struct drbd_peer_request
3684  */
drbd_wait_misc(struct drbd_device * device,struct drbd_interval * i)3685 int drbd_wait_misc(struct drbd_device *device, struct drbd_interval *i)
3686 {
3687 	struct net_conf *nc;
3688 	DEFINE_WAIT(wait);
3689 	long timeout;
3690 
3691 	rcu_read_lock();
3692 	nc = rcu_dereference(first_peer_device(device)->connection->net_conf);
3693 	if (!nc) {
3694 		rcu_read_unlock();
3695 		return -ETIMEDOUT;
3696 	}
3697 	timeout = nc->ko_count ? nc->timeout * HZ / 10 * nc->ko_count : MAX_SCHEDULE_TIMEOUT;
3698 	rcu_read_unlock();
3699 
3700 	/* Indicate to wake up device->misc_wait on progress.  */
3701 	i->waiting = true;
3702 	prepare_to_wait(&device->misc_wait, &wait, TASK_INTERRUPTIBLE);
3703 	spin_unlock_irq(&device->resource->req_lock);
3704 	timeout = schedule_timeout(timeout);
3705 	finish_wait(&device->misc_wait, &wait);
3706 	spin_lock_irq(&device->resource->req_lock);
3707 	if (!timeout || device->state.conn < C_CONNECTED)
3708 		return -ETIMEDOUT;
3709 	if (signal_pending(current))
3710 		return -ERESTARTSYS;
3711 	return 0;
3712 }
3713 
lock_all_resources(void)3714 void lock_all_resources(void)
3715 {
3716 	struct drbd_resource *resource;
3717 	int __maybe_unused i = 0;
3718 
3719 	mutex_lock(&resources_mutex);
3720 	local_irq_disable();
3721 	for_each_resource(resource, &drbd_resources)
3722 		spin_lock_nested(&resource->req_lock, i++);
3723 }
3724 
unlock_all_resources(void)3725 void unlock_all_resources(void)
3726 {
3727 	struct drbd_resource *resource;
3728 
3729 	for_each_resource(resource, &drbd_resources)
3730 		spin_unlock(&resource->req_lock);
3731 	local_irq_enable();
3732 	mutex_unlock(&resources_mutex);
3733 }
3734 
3735 #ifdef CONFIG_DRBD_FAULT_INJECTION
3736 /* Fault insertion support including random number generator shamelessly
3737  * stolen from kernel/rcutorture.c */
3738 struct fault_random_state {
3739 	unsigned long state;
3740 	unsigned long count;
3741 };
3742 
3743 #define FAULT_RANDOM_MULT 39916801  /* prime */
3744 #define FAULT_RANDOM_ADD	479001701 /* prime */
3745 #define FAULT_RANDOM_REFRESH 10000
3746 
3747 /*
3748  * Crude but fast random-number generator.  Uses a linear congruential
3749  * generator, with occasional help from get_random_bytes().
3750  */
3751 static unsigned long
_drbd_fault_random(struct fault_random_state * rsp)3752 _drbd_fault_random(struct fault_random_state *rsp)
3753 {
3754 	long refresh;
3755 
3756 	if (!rsp->count--) {
3757 		get_random_bytes(&refresh, sizeof(refresh));
3758 		rsp->state += refresh;
3759 		rsp->count = FAULT_RANDOM_REFRESH;
3760 	}
3761 	rsp->state = rsp->state * FAULT_RANDOM_MULT + FAULT_RANDOM_ADD;
3762 	return swahw32(rsp->state);
3763 }
3764 
3765 static char *
_drbd_fault_str(unsigned int type)3766 _drbd_fault_str(unsigned int type) {
3767 	static char *_faults[] = {
3768 		[DRBD_FAULT_MD_WR] = "Meta-data write",
3769 		[DRBD_FAULT_MD_RD] = "Meta-data read",
3770 		[DRBD_FAULT_RS_WR] = "Resync write",
3771 		[DRBD_FAULT_RS_RD] = "Resync read",
3772 		[DRBD_FAULT_DT_WR] = "Data write",
3773 		[DRBD_FAULT_DT_RD] = "Data read",
3774 		[DRBD_FAULT_DT_RA] = "Data read ahead",
3775 		[DRBD_FAULT_BM_ALLOC] = "BM allocation",
3776 		[DRBD_FAULT_AL_EE] = "EE allocation",
3777 		[DRBD_FAULT_RECEIVE] = "receive data corruption",
3778 	};
3779 
3780 	return (type < DRBD_FAULT_MAX) ? _faults[type] : "**Unknown**";
3781 }
3782 
3783 unsigned int
_drbd_insert_fault(struct drbd_device * device,unsigned int type)3784 _drbd_insert_fault(struct drbd_device *device, unsigned int type)
3785 {
3786 	static struct fault_random_state rrs = {0, 0};
3787 
3788 	unsigned int ret = (
3789 		(drbd_fault_devs == 0 ||
3790 			((1 << device_to_minor(device)) & drbd_fault_devs) != 0) &&
3791 		(((_drbd_fault_random(&rrs) % 100) + 1) <= drbd_fault_rate));
3792 
3793 	if (ret) {
3794 		drbd_fault_count++;
3795 
3796 		if (__ratelimit(&drbd_ratelimit_state))
3797 			drbd_warn(device, "***Simulating %s failure\n",
3798 				_drbd_fault_str(type));
3799 	}
3800 
3801 	return ret;
3802 }
3803 #endif
3804 
drbd_buildtag(void)3805 const char *drbd_buildtag(void)
3806 {
3807 	/* DRBD built from external sources has here a reference to the
3808 	   git hash of the source code. */
3809 
3810 	static char buildtag[38] = "\0uilt-in";
3811 
3812 	if (buildtag[0] == 0) {
3813 #ifdef MODULE
3814 		sprintf(buildtag, "srcversion: %-24s", THIS_MODULE->srcversion);
3815 #else
3816 		buildtag[0] = 'b';
3817 #endif
3818 	}
3819 
3820 	return buildtag;
3821 }
3822 
3823 module_init(drbd_init)
3824 module_exit(drbd_cleanup)
3825 
3826 EXPORT_SYMBOL(drbd_conn_str);
3827 EXPORT_SYMBOL(drbd_role_str);
3828 EXPORT_SYMBOL(drbd_disk_str);
3829 EXPORT_SYMBOL(drbd_set_st_err_str);
3830