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