1 /*
2  * VMware VMCI Driver
3  *
4  * Copyright (C) 2012 VMware, Inc. All rights reserved.
5  *
6  * This program is free software; you can redistribute it and/or modify it
7  * under the terms of the GNU General Public License as published by the
8  * Free Software Foundation version 2 and no later version.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
12  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
13  * for more details.
14  */
15 
16 #include <linux/vmw_vmci_defs.h>
17 #include <linux/vmw_vmci_api.h>
18 #include <linux/highmem.h>
19 #include <linux/kernel.h>
20 #include <linux/mm.h>
21 #include <linux/module.h>
22 #include <linux/mutex.h>
23 #include <linux/pagemap.h>
24 #include <linux/pci.h>
25 #include <linux/sched.h>
26 #include <linux/slab.h>
27 #include <linux/uio.h>
28 #include <linux/wait.h>
29 #include <linux/vmalloc.h>
30 #include <linux/skbuff.h>
31 
32 #include "vmci_handle_array.h"
33 #include "vmci_queue_pair.h"
34 #include "vmci_datagram.h"
35 #include "vmci_resource.h"
36 #include "vmci_context.h"
37 #include "vmci_driver.h"
38 #include "vmci_event.h"
39 #include "vmci_route.h"
40 
41 /*
42  * In the following, we will distinguish between two kinds of VMX processes -
43  * the ones with versions lower than VMCI_VERSION_NOVMVM that use specialized
44  * VMCI page files in the VMX and supporting VM to VM communication and the
45  * newer ones that use the guest memory directly. We will in the following
46  * refer to the older VMX versions as old-style VMX'en, and the newer ones as
47  * new-style VMX'en.
48  *
49  * The state transition datagram is as follows (the VMCIQPB_ prefix has been
50  * removed for readability) - see below for more details on the transtions:
51  *
52  *            --------------  NEW  -------------
53  *            |                                |
54  *           \_/                              \_/
55  *     CREATED_NO_MEM <-----------------> CREATED_MEM
56  *            |    |                           |
57  *            |    o-----------------------o   |
58  *            |                            |   |
59  *           \_/                          \_/ \_/
60  *     ATTACHED_NO_MEM <----------------> ATTACHED_MEM
61  *            |                            |   |
62  *            |     o----------------------o   |
63  *            |     |                          |
64  *           \_/   \_/                        \_/
65  *     SHUTDOWN_NO_MEM <----------------> SHUTDOWN_MEM
66  *            |                                |
67  *            |                                |
68  *            -------------> gone <-------------
69  *
70  * In more detail. When a VMCI queue pair is first created, it will be in the
71  * VMCIQPB_NEW state. It will then move into one of the following states:
72  *
73  * - VMCIQPB_CREATED_NO_MEM: this state indicates that either:
74  *
75  *     - the created was performed by a host endpoint, in which case there is
76  *       no backing memory yet.
77  *
78  *     - the create was initiated by an old-style VMX, that uses
79  *       vmci_qp_broker_set_page_store to specify the UVAs of the queue pair at
80  *       a later point in time. This state can be distinguished from the one
81  *       above by the context ID of the creator. A host side is not allowed to
82  *       attach until the page store has been set.
83  *
84  * - VMCIQPB_CREATED_MEM: this state is the result when the queue pair
85  *     is created by a VMX using the queue pair device backend that
86  *     sets the UVAs of the queue pair immediately and stores the
87  *     information for later attachers. At this point, it is ready for
88  *     the host side to attach to it.
89  *
90  * Once the queue pair is in one of the created states (with the exception of
91  * the case mentioned for older VMX'en above), it is possible to attach to the
92  * queue pair. Again we have two new states possible:
93  *
94  * - VMCIQPB_ATTACHED_MEM: this state can be reached through the following
95  *   paths:
96  *
97  *     - from VMCIQPB_CREATED_NO_MEM when a new-style VMX allocates a queue
98  *       pair, and attaches to a queue pair previously created by the host side.
99  *
100  *     - from VMCIQPB_CREATED_MEM when the host side attaches to a queue pair
101  *       already created by a guest.
102  *
103  *     - from VMCIQPB_ATTACHED_NO_MEM, when an old-style VMX calls
104  *       vmci_qp_broker_set_page_store (see below).
105  *
106  * - VMCIQPB_ATTACHED_NO_MEM: If the queue pair already was in the
107  *     VMCIQPB_CREATED_NO_MEM due to a host side create, an old-style VMX will
108  *     bring the queue pair into this state. Once vmci_qp_broker_set_page_store
109  *     is called to register the user memory, the VMCIQPB_ATTACH_MEM state
110  *     will be entered.
111  *
112  * From the attached queue pair, the queue pair can enter the shutdown states
113  * when either side of the queue pair detaches. If the guest side detaches
114  * first, the queue pair will enter the VMCIQPB_SHUTDOWN_NO_MEM state, where
115  * the content of the queue pair will no longer be available. If the host
116  * side detaches first, the queue pair will either enter the
117  * VMCIQPB_SHUTDOWN_MEM, if the guest memory is currently mapped, or
118  * VMCIQPB_SHUTDOWN_NO_MEM, if the guest memory is not mapped
119  * (e.g., the host detaches while a guest is stunned).
120  *
121  * New-style VMX'en will also unmap guest memory, if the guest is
122  * quiesced, e.g., during a snapshot operation. In that case, the guest
123  * memory will no longer be available, and the queue pair will transition from
124  * *_MEM state to a *_NO_MEM state. The VMX may later map the memory once more,
125  * in which case the queue pair will transition from the *_NO_MEM state at that
126  * point back to the *_MEM state. Note that the *_NO_MEM state may have changed,
127  * since the peer may have either attached or detached in the meantime. The
128  * values are laid out such that ++ on a state will move from a *_NO_MEM to a
129  * *_MEM state, and vice versa.
130  */
131 
132 /*
133  * VMCIMemcpy{To,From}QueueFunc() prototypes.  Functions of these
134  * types are passed around to enqueue and dequeue routines.  Note that
135  * often the functions passed are simply wrappers around memcpy
136  * itself.
137  *
138  * Note: In order for the memcpy typedefs to be compatible with the VMKernel,
139  * there's an unused last parameter for the hosted side.  In
140  * ESX, that parameter holds a buffer type.
141  */
142 typedef int vmci_memcpy_to_queue_func(struct vmci_queue *queue,
143 				      u64 queue_offset, const void *src,
144 				      size_t src_offset, size_t size);
145 typedef int vmci_memcpy_from_queue_func(void *dest, size_t dest_offset,
146 					const struct vmci_queue *queue,
147 					u64 queue_offset, size_t size);
148 
149 /* The Kernel specific component of the struct vmci_queue structure. */
150 struct vmci_queue_kern_if {
151 	struct mutex __mutex;	/* Protects the queue. */
152 	struct mutex *mutex;	/* Shared by producer and consumer queues. */
153 	size_t num_pages;	/* Number of pages incl. header. */
154 	bool host;		/* Host or guest? */
155 	union {
156 		struct {
157 			dma_addr_t *pas;
158 			void **vas;
159 		} g;		/* Used by the guest. */
160 		struct {
161 			struct page **page;
162 			struct page **header_page;
163 		} h;		/* Used by the host. */
164 	} u;
165 };
166 
167 /*
168  * This structure is opaque to the clients.
169  */
170 struct vmci_qp {
171 	struct vmci_handle handle;
172 	struct vmci_queue *produce_q;
173 	struct vmci_queue *consume_q;
174 	u64 produce_q_size;
175 	u64 consume_q_size;
176 	u32 peer;
177 	u32 flags;
178 	u32 priv_flags;
179 	bool guest_endpoint;
180 	unsigned int blocked;
181 	unsigned int generation;
182 	wait_queue_head_t event;
183 };
184 
185 enum qp_broker_state {
186 	VMCIQPB_NEW,
187 	VMCIQPB_CREATED_NO_MEM,
188 	VMCIQPB_CREATED_MEM,
189 	VMCIQPB_ATTACHED_NO_MEM,
190 	VMCIQPB_ATTACHED_MEM,
191 	VMCIQPB_SHUTDOWN_NO_MEM,
192 	VMCIQPB_SHUTDOWN_MEM,
193 	VMCIQPB_GONE
194 };
195 
196 #define QPBROKERSTATE_HAS_MEM(_qpb) (_qpb->state == VMCIQPB_CREATED_MEM || \
197 				     _qpb->state == VMCIQPB_ATTACHED_MEM || \
198 				     _qpb->state == VMCIQPB_SHUTDOWN_MEM)
199 
200 /*
201  * In the queue pair broker, we always use the guest point of view for
202  * the produce and consume queue values and references, e.g., the
203  * produce queue size stored is the guests produce queue size. The
204  * host endpoint will need to swap these around. The only exception is
205  * the local queue pairs on the host, in which case the host endpoint
206  * that creates the queue pair will have the right orientation, and
207  * the attaching host endpoint will need to swap.
208  */
209 struct qp_entry {
210 	struct list_head list_item;
211 	struct vmci_handle handle;
212 	u32 peer;
213 	u32 flags;
214 	u64 produce_size;
215 	u64 consume_size;
216 	u32 ref_count;
217 };
218 
219 struct qp_broker_entry {
220 	struct vmci_resource resource;
221 	struct qp_entry qp;
222 	u32 create_id;
223 	u32 attach_id;
224 	enum qp_broker_state state;
225 	bool require_trusted_attach;
226 	bool created_by_trusted;
227 	bool vmci_page_files;	/* Created by VMX using VMCI page files */
228 	struct vmci_queue *produce_q;
229 	struct vmci_queue *consume_q;
230 	struct vmci_queue_header saved_produce_q;
231 	struct vmci_queue_header saved_consume_q;
232 	vmci_event_release_cb wakeup_cb;
233 	void *client_data;
234 	void *local_mem;	/* Kernel memory for local queue pair */
235 };
236 
237 struct qp_guest_endpoint {
238 	struct vmci_resource resource;
239 	struct qp_entry qp;
240 	u64 num_ppns;
241 	void *produce_q;
242 	void *consume_q;
243 	struct ppn_set ppn_set;
244 };
245 
246 struct qp_list {
247 	struct list_head head;
248 	struct mutex mutex;	/* Protect queue list. */
249 };
250 
251 static struct qp_list qp_broker_list = {
252 	.head = LIST_HEAD_INIT(qp_broker_list.head),
253 	.mutex = __MUTEX_INITIALIZER(qp_broker_list.mutex),
254 };
255 
256 static struct qp_list qp_guest_endpoints = {
257 	.head = LIST_HEAD_INIT(qp_guest_endpoints.head),
258 	.mutex = __MUTEX_INITIALIZER(qp_guest_endpoints.mutex),
259 };
260 
261 #define INVALID_VMCI_GUEST_MEM_ID  0
262 #define QPE_NUM_PAGES(_QPE) ((u32) \
263 			     (DIV_ROUND_UP(_QPE.produce_size, PAGE_SIZE) + \
264 			      DIV_ROUND_UP(_QPE.consume_size, PAGE_SIZE) + 2))
265 
266 
267 /*
268  * Frees kernel VA space for a given queue and its queue header, and
269  * frees physical data pages.
270  */
271 static void qp_free_queue(void *q, u64 size)
272 {
273 	struct vmci_queue *queue = q;
274 
275 	if (queue) {
276 		u64 i;
277 
278 		/* Given size does not include header, so add in a page here. */
279 		for (i = 0; i < DIV_ROUND_UP(size, PAGE_SIZE) + 1; i++) {
280 			dma_free_coherent(&vmci_pdev->dev, PAGE_SIZE,
281 					  queue->kernel_if->u.g.vas[i],
282 					  queue->kernel_if->u.g.pas[i]);
283 		}
284 
285 		vfree(queue);
286 	}
287 }
288 
289 /*
290  * Allocates kernel queue pages of specified size with IOMMU mappings,
291  * plus space for the queue structure/kernel interface and the queue
292  * header.
293  */
294 static void *qp_alloc_queue(u64 size, u32 flags)
295 {
296 	u64 i;
297 	struct vmci_queue *queue;
298 	size_t pas_size;
299 	size_t vas_size;
300 	size_t queue_size = sizeof(*queue) + sizeof(*queue->kernel_if);
301 	const u64 num_pages = DIV_ROUND_UP(size, PAGE_SIZE) + 1;
302 
303 	if (num_pages >
304 		 (SIZE_MAX - queue_size) /
305 		 (sizeof(*queue->kernel_if->u.g.pas) +
306 		  sizeof(*queue->kernel_if->u.g.vas)))
307 		return NULL;
308 
309 	pas_size = num_pages * sizeof(*queue->kernel_if->u.g.pas);
310 	vas_size = num_pages * sizeof(*queue->kernel_if->u.g.vas);
311 	queue_size += pas_size + vas_size;
312 
313 	queue = vmalloc(queue_size);
314 	if (!queue)
315 		return NULL;
316 
317 	queue->q_header = NULL;
318 	queue->saved_header = NULL;
319 	queue->kernel_if = (struct vmci_queue_kern_if *)(queue + 1);
320 	queue->kernel_if->mutex = NULL;
321 	queue->kernel_if->num_pages = num_pages;
322 	queue->kernel_if->u.g.pas = (dma_addr_t *)(queue->kernel_if + 1);
323 	queue->kernel_if->u.g.vas =
324 		(void **)((u8 *)queue->kernel_if->u.g.pas + pas_size);
325 	queue->kernel_if->host = false;
326 
327 	for (i = 0; i < num_pages; i++) {
328 		queue->kernel_if->u.g.vas[i] =
329 			dma_alloc_coherent(&vmci_pdev->dev, PAGE_SIZE,
330 					   &queue->kernel_if->u.g.pas[i],
331 					   GFP_KERNEL);
332 		if (!queue->kernel_if->u.g.vas[i]) {
333 			/* Size excl. the header. */
334 			qp_free_queue(queue, i * PAGE_SIZE);
335 			return NULL;
336 		}
337 	}
338 
339 	/* Queue header is the first page. */
340 	queue->q_header = queue->kernel_if->u.g.vas[0];
341 
342 	return queue;
343 }
344 
345 /*
346  * Copies from a given buffer or iovector to a VMCI Queue.  Uses
347  * kmap()/kunmap() to dynamically map/unmap required portions of the queue
348  * by traversing the offset -> page translation structure for the queue.
349  * Assumes that offset + size does not wrap around in the queue.
350  */
351 static int __qp_memcpy_to_queue(struct vmci_queue *queue,
352 				u64 queue_offset,
353 				const void *src,
354 				size_t size,
355 				bool is_iovec)
356 {
357 	struct vmci_queue_kern_if *kernel_if = queue->kernel_if;
358 	size_t bytes_copied = 0;
359 
360 	while (bytes_copied < size) {
361 		const u64 page_index =
362 			(queue_offset + bytes_copied) / PAGE_SIZE;
363 		const size_t page_offset =
364 		    (queue_offset + bytes_copied) & (PAGE_SIZE - 1);
365 		void *va;
366 		size_t to_copy;
367 
368 		if (kernel_if->host)
369 			va = kmap(kernel_if->u.h.page[page_index]);
370 		else
371 			va = kernel_if->u.g.vas[page_index + 1];
372 			/* Skip header. */
373 
374 		if (size - bytes_copied > PAGE_SIZE - page_offset)
375 			/* Enough payload to fill up from this page. */
376 			to_copy = PAGE_SIZE - page_offset;
377 		else
378 			to_copy = size - bytes_copied;
379 
380 		if (is_iovec) {
381 			struct msghdr *msg = (struct msghdr *)src;
382 			int err;
383 
384 			/* The iovec will track bytes_copied internally. */
385 			err = memcpy_from_msg((u8 *)va + page_offset,
386 					      msg, to_copy);
387 			if (err != 0) {
388 				if (kernel_if->host)
389 					kunmap(kernel_if->u.h.page[page_index]);
390 				return VMCI_ERROR_INVALID_ARGS;
391 			}
392 		} else {
393 			memcpy((u8 *)va + page_offset,
394 			       (u8 *)src + bytes_copied, to_copy);
395 		}
396 
397 		bytes_copied += to_copy;
398 		if (kernel_if->host)
399 			kunmap(kernel_if->u.h.page[page_index]);
400 	}
401 
402 	return VMCI_SUCCESS;
403 }
404 
405 /*
406  * Copies to a given buffer or iovector from a VMCI Queue.  Uses
407  * kmap()/kunmap() to dynamically map/unmap required portions of the queue
408  * by traversing the offset -> page translation structure for the queue.
409  * Assumes that offset + size does not wrap around in the queue.
410  */
411 static int __qp_memcpy_from_queue(void *dest,
412 				  const struct vmci_queue *queue,
413 				  u64 queue_offset,
414 				  size_t size,
415 				  bool is_iovec)
416 {
417 	struct vmci_queue_kern_if *kernel_if = queue->kernel_if;
418 	size_t bytes_copied = 0;
419 
420 	while (bytes_copied < size) {
421 		const u64 page_index =
422 			(queue_offset + bytes_copied) / PAGE_SIZE;
423 		const size_t page_offset =
424 		    (queue_offset + bytes_copied) & (PAGE_SIZE - 1);
425 		void *va;
426 		size_t to_copy;
427 
428 		if (kernel_if->host)
429 			va = kmap(kernel_if->u.h.page[page_index]);
430 		else
431 			va = kernel_if->u.g.vas[page_index + 1];
432 			/* Skip header. */
433 
434 		if (size - bytes_copied > PAGE_SIZE - page_offset)
435 			/* Enough payload to fill up this page. */
436 			to_copy = PAGE_SIZE - page_offset;
437 		else
438 			to_copy = size - bytes_copied;
439 
440 		if (is_iovec) {
441 			struct msghdr *msg = dest;
442 			int err;
443 
444 			/* The iovec will track bytes_copied internally. */
445 			err = memcpy_to_msg(msg, (u8 *)va + page_offset,
446 					     to_copy);
447 			if (err != 0) {
448 				if (kernel_if->host)
449 					kunmap(kernel_if->u.h.page[page_index]);
450 				return VMCI_ERROR_INVALID_ARGS;
451 			}
452 		} else {
453 			memcpy((u8 *)dest + bytes_copied,
454 			       (u8 *)va + page_offset, to_copy);
455 		}
456 
457 		bytes_copied += to_copy;
458 		if (kernel_if->host)
459 			kunmap(kernel_if->u.h.page[page_index]);
460 	}
461 
462 	return VMCI_SUCCESS;
463 }
464 
465 /*
466  * Allocates two list of PPNs --- one for the pages in the produce queue,
467  * and the other for the pages in the consume queue. Intializes the list
468  * of PPNs with the page frame numbers of the KVA for the two queues (and
469  * the queue headers).
470  */
471 static int qp_alloc_ppn_set(void *prod_q,
472 			    u64 num_produce_pages,
473 			    void *cons_q,
474 			    u64 num_consume_pages, struct ppn_set *ppn_set)
475 {
476 	u32 *produce_ppns;
477 	u32 *consume_ppns;
478 	struct vmci_queue *produce_q = prod_q;
479 	struct vmci_queue *consume_q = cons_q;
480 	u64 i;
481 
482 	if (!produce_q || !num_produce_pages || !consume_q ||
483 	    !num_consume_pages || !ppn_set)
484 		return VMCI_ERROR_INVALID_ARGS;
485 
486 	if (ppn_set->initialized)
487 		return VMCI_ERROR_ALREADY_EXISTS;
488 
489 	produce_ppns =
490 	    kmalloc(num_produce_pages * sizeof(*produce_ppns), GFP_KERNEL);
491 	if (!produce_ppns)
492 		return VMCI_ERROR_NO_MEM;
493 
494 	consume_ppns =
495 	    kmalloc(num_consume_pages * sizeof(*consume_ppns), GFP_KERNEL);
496 	if (!consume_ppns) {
497 		kfree(produce_ppns);
498 		return VMCI_ERROR_NO_MEM;
499 	}
500 
501 	for (i = 0; i < num_produce_pages; i++) {
502 		unsigned long pfn;
503 
504 		produce_ppns[i] =
505 			produce_q->kernel_if->u.g.pas[i] >> PAGE_SHIFT;
506 		pfn = produce_ppns[i];
507 
508 		/* Fail allocation if PFN isn't supported by hypervisor. */
509 		if (sizeof(pfn) > sizeof(*produce_ppns)
510 		    && pfn != produce_ppns[i])
511 			goto ppn_error;
512 	}
513 
514 	for (i = 0; i < num_consume_pages; i++) {
515 		unsigned long pfn;
516 
517 		consume_ppns[i] =
518 			consume_q->kernel_if->u.g.pas[i] >> PAGE_SHIFT;
519 		pfn = consume_ppns[i];
520 
521 		/* Fail allocation if PFN isn't supported by hypervisor. */
522 		if (sizeof(pfn) > sizeof(*consume_ppns)
523 		    && pfn != consume_ppns[i])
524 			goto ppn_error;
525 	}
526 
527 	ppn_set->num_produce_pages = num_produce_pages;
528 	ppn_set->num_consume_pages = num_consume_pages;
529 	ppn_set->produce_ppns = produce_ppns;
530 	ppn_set->consume_ppns = consume_ppns;
531 	ppn_set->initialized = true;
532 	return VMCI_SUCCESS;
533 
534  ppn_error:
535 	kfree(produce_ppns);
536 	kfree(consume_ppns);
537 	return VMCI_ERROR_INVALID_ARGS;
538 }
539 
540 /*
541  * Frees the two list of PPNs for a queue pair.
542  */
543 static void qp_free_ppn_set(struct ppn_set *ppn_set)
544 {
545 	if (ppn_set->initialized) {
546 		/* Do not call these functions on NULL inputs. */
547 		kfree(ppn_set->produce_ppns);
548 		kfree(ppn_set->consume_ppns);
549 	}
550 	memset(ppn_set, 0, sizeof(*ppn_set));
551 }
552 
553 /*
554  * Populates the list of PPNs in the hypercall structure with the PPNS
555  * of the produce queue and the consume queue.
556  */
557 static int qp_populate_ppn_set(u8 *call_buf, const struct ppn_set *ppn_set)
558 {
559 	memcpy(call_buf, ppn_set->produce_ppns,
560 	       ppn_set->num_produce_pages * sizeof(*ppn_set->produce_ppns));
561 	memcpy(call_buf +
562 	       ppn_set->num_produce_pages * sizeof(*ppn_set->produce_ppns),
563 	       ppn_set->consume_ppns,
564 	       ppn_set->num_consume_pages * sizeof(*ppn_set->consume_ppns));
565 
566 	return VMCI_SUCCESS;
567 }
568 
569 static int qp_memcpy_to_queue(struct vmci_queue *queue,
570 			      u64 queue_offset,
571 			      const void *src, size_t src_offset, size_t size)
572 {
573 	return __qp_memcpy_to_queue(queue, queue_offset,
574 				    (u8 *)src + src_offset, size, false);
575 }
576 
577 static int qp_memcpy_from_queue(void *dest,
578 				size_t dest_offset,
579 				const struct vmci_queue *queue,
580 				u64 queue_offset, size_t size)
581 {
582 	return __qp_memcpy_from_queue((u8 *)dest + dest_offset,
583 				      queue, queue_offset, size, false);
584 }
585 
586 /*
587  * Copies from a given iovec from a VMCI Queue.
588  */
589 static int qp_memcpy_to_queue_iov(struct vmci_queue *queue,
590 				  u64 queue_offset,
591 				  const void *msg,
592 				  size_t src_offset, size_t size)
593 {
594 
595 	/*
596 	 * We ignore src_offset because src is really a struct iovec * and will
597 	 * maintain offset internally.
598 	 */
599 	return __qp_memcpy_to_queue(queue, queue_offset, msg, size, true);
600 }
601 
602 /*
603  * Copies to a given iovec from a VMCI Queue.
604  */
605 static int qp_memcpy_from_queue_iov(void *dest,
606 				    size_t dest_offset,
607 				    const struct vmci_queue *queue,
608 				    u64 queue_offset, size_t size)
609 {
610 	/*
611 	 * We ignore dest_offset because dest is really a struct iovec * and
612 	 * will maintain offset internally.
613 	 */
614 	return __qp_memcpy_from_queue(dest, queue, queue_offset, size, true);
615 }
616 
617 /*
618  * Allocates kernel VA space of specified size plus space for the queue
619  * and kernel interface.  This is different from the guest queue allocator,
620  * because we do not allocate our own queue header/data pages here but
621  * share those of the guest.
622  */
623 static struct vmci_queue *qp_host_alloc_queue(u64 size)
624 {
625 	struct vmci_queue *queue;
626 	size_t queue_page_size;
627 	const u64 num_pages = DIV_ROUND_UP(size, PAGE_SIZE) + 1;
628 	const size_t queue_size = sizeof(*queue) + sizeof(*(queue->kernel_if));
629 
630 	if (num_pages > (SIZE_MAX - queue_size) /
631 		 sizeof(*queue->kernel_if->u.h.page))
632 		return NULL;
633 
634 	queue_page_size = num_pages * sizeof(*queue->kernel_if->u.h.page);
635 
636 	queue = kzalloc(queue_size + queue_page_size, GFP_KERNEL);
637 	if (queue) {
638 		queue->q_header = NULL;
639 		queue->saved_header = NULL;
640 		queue->kernel_if = (struct vmci_queue_kern_if *)(queue + 1);
641 		queue->kernel_if->host = true;
642 		queue->kernel_if->mutex = NULL;
643 		queue->kernel_if->num_pages = num_pages;
644 		queue->kernel_if->u.h.header_page =
645 		    (struct page **)((u8 *)queue + queue_size);
646 		queue->kernel_if->u.h.page =
647 			&queue->kernel_if->u.h.header_page[1];
648 	}
649 
650 	return queue;
651 }
652 
653 /*
654  * Frees kernel memory for a given queue (header plus translation
655  * structure).
656  */
657 static void qp_host_free_queue(struct vmci_queue *queue, u64 queue_size)
658 {
659 	kfree(queue);
660 }
661 
662 /*
663  * Initialize the mutex for the pair of queues.  This mutex is used to
664  * protect the q_header and the buffer from changing out from under any
665  * users of either queue.  Of course, it's only any good if the mutexes
666  * are actually acquired.  Queue structure must lie on non-paged memory
667  * or we cannot guarantee access to the mutex.
668  */
669 static void qp_init_queue_mutex(struct vmci_queue *produce_q,
670 				struct vmci_queue *consume_q)
671 {
672 	/*
673 	 * Only the host queue has shared state - the guest queues do not
674 	 * need to synchronize access using a queue mutex.
675 	 */
676 
677 	if (produce_q->kernel_if->host) {
678 		produce_q->kernel_if->mutex = &produce_q->kernel_if->__mutex;
679 		consume_q->kernel_if->mutex = &produce_q->kernel_if->__mutex;
680 		mutex_init(produce_q->kernel_if->mutex);
681 	}
682 }
683 
684 /*
685  * Cleans up the mutex for the pair of queues.
686  */
687 static void qp_cleanup_queue_mutex(struct vmci_queue *produce_q,
688 				   struct vmci_queue *consume_q)
689 {
690 	if (produce_q->kernel_if->host) {
691 		produce_q->kernel_if->mutex = NULL;
692 		consume_q->kernel_if->mutex = NULL;
693 	}
694 }
695 
696 /*
697  * Acquire the mutex for the queue.  Note that the produce_q and
698  * the consume_q share a mutex.  So, only one of the two need to
699  * be passed in to this routine.  Either will work just fine.
700  */
701 static void qp_acquire_queue_mutex(struct vmci_queue *queue)
702 {
703 	if (queue->kernel_if->host)
704 		mutex_lock(queue->kernel_if->mutex);
705 }
706 
707 /*
708  * Release the mutex for the queue.  Note that the produce_q and
709  * the consume_q share a mutex.  So, only one of the two need to
710  * be passed in to this routine.  Either will work just fine.
711  */
712 static void qp_release_queue_mutex(struct vmci_queue *queue)
713 {
714 	if (queue->kernel_if->host)
715 		mutex_unlock(queue->kernel_if->mutex);
716 }
717 
718 /*
719  * Helper function to release pages in the PageStoreAttachInfo
720  * previously obtained using get_user_pages.
721  */
722 static void qp_release_pages(struct page **pages,
723 			     u64 num_pages, bool dirty)
724 {
725 	int i;
726 
727 	for (i = 0; i < num_pages; i++) {
728 		if (dirty)
729 			set_page_dirty(pages[i]);
730 
731 		page_cache_release(pages[i]);
732 		pages[i] = NULL;
733 	}
734 }
735 
736 /*
737  * Lock the user pages referenced by the {produce,consume}Buffer
738  * struct into memory and populate the {produce,consume}Pages
739  * arrays in the attach structure with them.
740  */
741 static int qp_host_get_user_memory(u64 produce_uva,
742 				   u64 consume_uva,
743 				   struct vmci_queue *produce_q,
744 				   struct vmci_queue *consume_q)
745 {
746 	int retval;
747 	int err = VMCI_SUCCESS;
748 
749 	retval = get_user_pages_fast((uintptr_t) produce_uva,
750 				     produce_q->kernel_if->num_pages, 1,
751 				     produce_q->kernel_if->u.h.header_page);
752 	if (retval < produce_q->kernel_if->num_pages) {
753 		pr_debug("get_user_pages_fast(produce) failed (retval=%d)",
754 			retval);
755 		qp_release_pages(produce_q->kernel_if->u.h.header_page,
756 				 retval, false);
757 		err = VMCI_ERROR_NO_MEM;
758 		goto out;
759 	}
760 
761 	retval = get_user_pages_fast((uintptr_t) consume_uva,
762 				     consume_q->kernel_if->num_pages, 1,
763 				     consume_q->kernel_if->u.h.header_page);
764 	if (retval < consume_q->kernel_if->num_pages) {
765 		pr_debug("get_user_pages_fast(consume) failed (retval=%d)",
766 			retval);
767 		qp_release_pages(consume_q->kernel_if->u.h.header_page,
768 				 retval, false);
769 		qp_release_pages(produce_q->kernel_if->u.h.header_page,
770 				 produce_q->kernel_if->num_pages, false);
771 		err = VMCI_ERROR_NO_MEM;
772 	}
773 
774  out:
775 	return err;
776 }
777 
778 /*
779  * Registers the specification of the user pages used for backing a queue
780  * pair. Enough information to map in pages is stored in the OS specific
781  * part of the struct vmci_queue structure.
782  */
783 static int qp_host_register_user_memory(struct vmci_qp_page_store *page_store,
784 					struct vmci_queue *produce_q,
785 					struct vmci_queue *consume_q)
786 {
787 	u64 produce_uva;
788 	u64 consume_uva;
789 
790 	/*
791 	 * The new style and the old style mapping only differs in
792 	 * that we either get a single or two UVAs, so we split the
793 	 * single UVA range at the appropriate spot.
794 	 */
795 	produce_uva = page_store->pages;
796 	consume_uva = page_store->pages +
797 	    produce_q->kernel_if->num_pages * PAGE_SIZE;
798 	return qp_host_get_user_memory(produce_uva, consume_uva, produce_q,
799 				       consume_q);
800 }
801 
802 /*
803  * Releases and removes the references to user pages stored in the attach
804  * struct.  Pages are released from the page cache and may become
805  * swappable again.
806  */
807 static void qp_host_unregister_user_memory(struct vmci_queue *produce_q,
808 					   struct vmci_queue *consume_q)
809 {
810 	qp_release_pages(produce_q->kernel_if->u.h.header_page,
811 			 produce_q->kernel_if->num_pages, true);
812 	memset(produce_q->kernel_if->u.h.header_page, 0,
813 	       sizeof(*produce_q->kernel_if->u.h.header_page) *
814 	       produce_q->kernel_if->num_pages);
815 	qp_release_pages(consume_q->kernel_if->u.h.header_page,
816 			 consume_q->kernel_if->num_pages, true);
817 	memset(consume_q->kernel_if->u.h.header_page, 0,
818 	       sizeof(*consume_q->kernel_if->u.h.header_page) *
819 	       consume_q->kernel_if->num_pages);
820 }
821 
822 /*
823  * Once qp_host_register_user_memory has been performed on a
824  * queue, the queue pair headers can be mapped into the
825  * kernel. Once mapped, they must be unmapped with
826  * qp_host_unmap_queues prior to calling
827  * qp_host_unregister_user_memory.
828  * Pages are pinned.
829  */
830 static int qp_host_map_queues(struct vmci_queue *produce_q,
831 			      struct vmci_queue *consume_q)
832 {
833 	int result;
834 
835 	if (!produce_q->q_header || !consume_q->q_header) {
836 		struct page *headers[2];
837 
838 		if (produce_q->q_header != consume_q->q_header)
839 			return VMCI_ERROR_QUEUEPAIR_MISMATCH;
840 
841 		if (produce_q->kernel_if->u.h.header_page == NULL ||
842 		    *produce_q->kernel_if->u.h.header_page == NULL)
843 			return VMCI_ERROR_UNAVAILABLE;
844 
845 		headers[0] = *produce_q->kernel_if->u.h.header_page;
846 		headers[1] = *consume_q->kernel_if->u.h.header_page;
847 
848 		produce_q->q_header = vmap(headers, 2, VM_MAP, PAGE_KERNEL);
849 		if (produce_q->q_header != NULL) {
850 			consume_q->q_header =
851 			    (struct vmci_queue_header *)((u8 *)
852 							 produce_q->q_header +
853 							 PAGE_SIZE);
854 			result = VMCI_SUCCESS;
855 		} else {
856 			pr_warn("vmap failed\n");
857 			result = VMCI_ERROR_NO_MEM;
858 		}
859 	} else {
860 		result = VMCI_SUCCESS;
861 	}
862 
863 	return result;
864 }
865 
866 /*
867  * Unmaps previously mapped queue pair headers from the kernel.
868  * Pages are unpinned.
869  */
870 static int qp_host_unmap_queues(u32 gid,
871 				struct vmci_queue *produce_q,
872 				struct vmci_queue *consume_q)
873 {
874 	if (produce_q->q_header) {
875 		if (produce_q->q_header < consume_q->q_header)
876 			vunmap(produce_q->q_header);
877 		else
878 			vunmap(consume_q->q_header);
879 
880 		produce_q->q_header = NULL;
881 		consume_q->q_header = NULL;
882 	}
883 
884 	return VMCI_SUCCESS;
885 }
886 
887 /*
888  * Finds the entry in the list corresponding to a given handle. Assumes
889  * that the list is locked.
890  */
891 static struct qp_entry *qp_list_find(struct qp_list *qp_list,
892 				     struct vmci_handle handle)
893 {
894 	struct qp_entry *entry;
895 
896 	if (vmci_handle_is_invalid(handle))
897 		return NULL;
898 
899 	list_for_each_entry(entry, &qp_list->head, list_item) {
900 		if (vmci_handle_is_equal(entry->handle, handle))
901 			return entry;
902 	}
903 
904 	return NULL;
905 }
906 
907 /*
908  * Finds the entry in the list corresponding to a given handle.
909  */
910 static struct qp_guest_endpoint *
911 qp_guest_handle_to_entry(struct vmci_handle handle)
912 {
913 	struct qp_guest_endpoint *entry;
914 	struct qp_entry *qp = qp_list_find(&qp_guest_endpoints, handle);
915 
916 	entry = qp ? container_of(
917 		qp, struct qp_guest_endpoint, qp) : NULL;
918 	return entry;
919 }
920 
921 /*
922  * Finds the entry in the list corresponding to a given handle.
923  */
924 static struct qp_broker_entry *
925 qp_broker_handle_to_entry(struct vmci_handle handle)
926 {
927 	struct qp_broker_entry *entry;
928 	struct qp_entry *qp = qp_list_find(&qp_broker_list, handle);
929 
930 	entry = qp ? container_of(
931 		qp, struct qp_broker_entry, qp) : NULL;
932 	return entry;
933 }
934 
935 /*
936  * Dispatches a queue pair event message directly into the local event
937  * queue.
938  */
939 static int qp_notify_peer_local(bool attach, struct vmci_handle handle)
940 {
941 	u32 context_id = vmci_get_context_id();
942 	struct vmci_event_qp ev;
943 
944 	ev.msg.hdr.dst = vmci_make_handle(context_id, VMCI_EVENT_HANDLER);
945 	ev.msg.hdr.src = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
946 					  VMCI_CONTEXT_RESOURCE_ID);
947 	ev.msg.hdr.payload_size = sizeof(ev) - sizeof(ev.msg.hdr);
948 	ev.msg.event_data.event =
949 	    attach ? VMCI_EVENT_QP_PEER_ATTACH : VMCI_EVENT_QP_PEER_DETACH;
950 	ev.payload.peer_id = context_id;
951 	ev.payload.handle = handle;
952 
953 	return vmci_event_dispatch(&ev.msg.hdr);
954 }
955 
956 /*
957  * Allocates and initializes a qp_guest_endpoint structure.
958  * Allocates a queue_pair rid (and handle) iff the given entry has
959  * an invalid handle.  0 through VMCI_RESERVED_RESOURCE_ID_MAX
960  * are reserved handles.  Assumes that the QP list mutex is held
961  * by the caller.
962  */
963 static struct qp_guest_endpoint *
964 qp_guest_endpoint_create(struct vmci_handle handle,
965 			 u32 peer,
966 			 u32 flags,
967 			 u64 produce_size,
968 			 u64 consume_size,
969 			 void *produce_q,
970 			 void *consume_q)
971 {
972 	int result;
973 	struct qp_guest_endpoint *entry;
974 	/* One page each for the queue headers. */
975 	const u64 num_ppns = DIV_ROUND_UP(produce_size, PAGE_SIZE) +
976 	    DIV_ROUND_UP(consume_size, PAGE_SIZE) + 2;
977 
978 	if (vmci_handle_is_invalid(handle)) {
979 		u32 context_id = vmci_get_context_id();
980 
981 		handle = vmci_make_handle(context_id, VMCI_INVALID_ID);
982 	}
983 
984 	entry = kzalloc(sizeof(*entry), GFP_KERNEL);
985 	if (entry) {
986 		entry->qp.peer = peer;
987 		entry->qp.flags = flags;
988 		entry->qp.produce_size = produce_size;
989 		entry->qp.consume_size = consume_size;
990 		entry->qp.ref_count = 0;
991 		entry->num_ppns = num_ppns;
992 		entry->produce_q = produce_q;
993 		entry->consume_q = consume_q;
994 		INIT_LIST_HEAD(&entry->qp.list_item);
995 
996 		/* Add resource obj */
997 		result = vmci_resource_add(&entry->resource,
998 					   VMCI_RESOURCE_TYPE_QPAIR_GUEST,
999 					   handle);
1000 		entry->qp.handle = vmci_resource_handle(&entry->resource);
1001 		if ((result != VMCI_SUCCESS) ||
1002 		    qp_list_find(&qp_guest_endpoints, entry->qp.handle)) {
1003 			pr_warn("Failed to add new resource (handle=0x%x:0x%x), error: %d",
1004 				handle.context, handle.resource, result);
1005 			kfree(entry);
1006 			entry = NULL;
1007 		}
1008 	}
1009 	return entry;
1010 }
1011 
1012 /*
1013  * Frees a qp_guest_endpoint structure.
1014  */
1015 static void qp_guest_endpoint_destroy(struct qp_guest_endpoint *entry)
1016 {
1017 	qp_free_ppn_set(&entry->ppn_set);
1018 	qp_cleanup_queue_mutex(entry->produce_q, entry->consume_q);
1019 	qp_free_queue(entry->produce_q, entry->qp.produce_size);
1020 	qp_free_queue(entry->consume_q, entry->qp.consume_size);
1021 	/* Unlink from resource hash table and free callback */
1022 	vmci_resource_remove(&entry->resource);
1023 
1024 	kfree(entry);
1025 }
1026 
1027 /*
1028  * Helper to make a queue_pairAlloc hypercall when the driver is
1029  * supporting a guest device.
1030  */
1031 static int qp_alloc_hypercall(const struct qp_guest_endpoint *entry)
1032 {
1033 	struct vmci_qp_alloc_msg *alloc_msg;
1034 	size_t msg_size;
1035 	int result;
1036 
1037 	if (!entry || entry->num_ppns <= 2)
1038 		return VMCI_ERROR_INVALID_ARGS;
1039 
1040 	msg_size = sizeof(*alloc_msg) +
1041 	    (size_t) entry->num_ppns * sizeof(u32);
1042 	alloc_msg = kmalloc(msg_size, GFP_KERNEL);
1043 	if (!alloc_msg)
1044 		return VMCI_ERROR_NO_MEM;
1045 
1046 	alloc_msg->hdr.dst = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
1047 					      VMCI_QUEUEPAIR_ALLOC);
1048 	alloc_msg->hdr.src = VMCI_ANON_SRC_HANDLE;
1049 	alloc_msg->hdr.payload_size = msg_size - VMCI_DG_HEADERSIZE;
1050 	alloc_msg->handle = entry->qp.handle;
1051 	alloc_msg->peer = entry->qp.peer;
1052 	alloc_msg->flags = entry->qp.flags;
1053 	alloc_msg->produce_size = entry->qp.produce_size;
1054 	alloc_msg->consume_size = entry->qp.consume_size;
1055 	alloc_msg->num_ppns = entry->num_ppns;
1056 
1057 	result = qp_populate_ppn_set((u8 *)alloc_msg + sizeof(*alloc_msg),
1058 				     &entry->ppn_set);
1059 	if (result == VMCI_SUCCESS)
1060 		result = vmci_send_datagram(&alloc_msg->hdr);
1061 
1062 	kfree(alloc_msg);
1063 
1064 	return result;
1065 }
1066 
1067 /*
1068  * Helper to make a queue_pairDetach hypercall when the driver is
1069  * supporting a guest device.
1070  */
1071 static int qp_detatch_hypercall(struct vmci_handle handle)
1072 {
1073 	struct vmci_qp_detach_msg detach_msg;
1074 
1075 	detach_msg.hdr.dst = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
1076 					      VMCI_QUEUEPAIR_DETACH);
1077 	detach_msg.hdr.src = VMCI_ANON_SRC_HANDLE;
1078 	detach_msg.hdr.payload_size = sizeof(handle);
1079 	detach_msg.handle = handle;
1080 
1081 	return vmci_send_datagram(&detach_msg.hdr);
1082 }
1083 
1084 /*
1085  * Adds the given entry to the list. Assumes that the list is locked.
1086  */
1087 static void qp_list_add_entry(struct qp_list *qp_list, struct qp_entry *entry)
1088 {
1089 	if (entry)
1090 		list_add(&entry->list_item, &qp_list->head);
1091 }
1092 
1093 /*
1094  * Removes the given entry from the list. Assumes that the list is locked.
1095  */
1096 static void qp_list_remove_entry(struct qp_list *qp_list,
1097 				 struct qp_entry *entry)
1098 {
1099 	if (entry)
1100 		list_del(&entry->list_item);
1101 }
1102 
1103 /*
1104  * Helper for VMCI queue_pair detach interface. Frees the physical
1105  * pages for the queue pair.
1106  */
1107 static int qp_detatch_guest_work(struct vmci_handle handle)
1108 {
1109 	int result;
1110 	struct qp_guest_endpoint *entry;
1111 	u32 ref_count = ~0;	/* To avoid compiler warning below */
1112 
1113 	mutex_lock(&qp_guest_endpoints.mutex);
1114 
1115 	entry = qp_guest_handle_to_entry(handle);
1116 	if (!entry) {
1117 		mutex_unlock(&qp_guest_endpoints.mutex);
1118 		return VMCI_ERROR_NOT_FOUND;
1119 	}
1120 
1121 	if (entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1122 		result = VMCI_SUCCESS;
1123 
1124 		if (entry->qp.ref_count > 1) {
1125 			result = qp_notify_peer_local(false, handle);
1126 			/*
1127 			 * We can fail to notify a local queuepair
1128 			 * because we can't allocate.  We still want
1129 			 * to release the entry if that happens, so
1130 			 * don't bail out yet.
1131 			 */
1132 		}
1133 	} else {
1134 		result = qp_detatch_hypercall(handle);
1135 		if (result < VMCI_SUCCESS) {
1136 			/*
1137 			 * We failed to notify a non-local queuepair.
1138 			 * That other queuepair might still be
1139 			 * accessing the shared memory, so don't
1140 			 * release the entry yet.  It will get cleaned
1141 			 * up by VMCIqueue_pair_Exit() if necessary
1142 			 * (assuming we are going away, otherwise why
1143 			 * did this fail?).
1144 			 */
1145 
1146 			mutex_unlock(&qp_guest_endpoints.mutex);
1147 			return result;
1148 		}
1149 	}
1150 
1151 	/*
1152 	 * If we get here then we either failed to notify a local queuepair, or
1153 	 * we succeeded in all cases.  Release the entry if required.
1154 	 */
1155 
1156 	entry->qp.ref_count--;
1157 	if (entry->qp.ref_count == 0)
1158 		qp_list_remove_entry(&qp_guest_endpoints, &entry->qp);
1159 
1160 	/* If we didn't remove the entry, this could change once we unlock. */
1161 	if (entry)
1162 		ref_count = entry->qp.ref_count;
1163 
1164 	mutex_unlock(&qp_guest_endpoints.mutex);
1165 
1166 	if (ref_count == 0)
1167 		qp_guest_endpoint_destroy(entry);
1168 
1169 	return result;
1170 }
1171 
1172 /*
1173  * This functions handles the actual allocation of a VMCI queue
1174  * pair guest endpoint. Allocates physical pages for the queue
1175  * pair. It makes OS dependent calls through generic wrappers.
1176  */
1177 static int qp_alloc_guest_work(struct vmci_handle *handle,
1178 			       struct vmci_queue **produce_q,
1179 			       u64 produce_size,
1180 			       struct vmci_queue **consume_q,
1181 			       u64 consume_size,
1182 			       u32 peer,
1183 			       u32 flags,
1184 			       u32 priv_flags)
1185 {
1186 	const u64 num_produce_pages =
1187 	    DIV_ROUND_UP(produce_size, PAGE_SIZE) + 1;
1188 	const u64 num_consume_pages =
1189 	    DIV_ROUND_UP(consume_size, PAGE_SIZE) + 1;
1190 	void *my_produce_q = NULL;
1191 	void *my_consume_q = NULL;
1192 	int result;
1193 	struct qp_guest_endpoint *queue_pair_entry = NULL;
1194 
1195 	if (priv_flags != VMCI_NO_PRIVILEGE_FLAGS)
1196 		return VMCI_ERROR_NO_ACCESS;
1197 
1198 	mutex_lock(&qp_guest_endpoints.mutex);
1199 
1200 	queue_pair_entry = qp_guest_handle_to_entry(*handle);
1201 	if (queue_pair_entry) {
1202 		if (queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1203 			/* Local attach case. */
1204 			if (queue_pair_entry->qp.ref_count > 1) {
1205 				pr_devel("Error attempting to attach more than once\n");
1206 				result = VMCI_ERROR_UNAVAILABLE;
1207 				goto error_keep_entry;
1208 			}
1209 
1210 			if (queue_pair_entry->qp.produce_size != consume_size ||
1211 			    queue_pair_entry->qp.consume_size !=
1212 			    produce_size ||
1213 			    queue_pair_entry->qp.flags !=
1214 			    (flags & ~VMCI_QPFLAG_ATTACH_ONLY)) {
1215 				pr_devel("Error mismatched queue pair in local attach\n");
1216 				result = VMCI_ERROR_QUEUEPAIR_MISMATCH;
1217 				goto error_keep_entry;
1218 			}
1219 
1220 			/*
1221 			 * Do a local attach.  We swap the consume and
1222 			 * produce queues for the attacher and deliver
1223 			 * an attach event.
1224 			 */
1225 			result = qp_notify_peer_local(true, *handle);
1226 			if (result < VMCI_SUCCESS)
1227 				goto error_keep_entry;
1228 
1229 			my_produce_q = queue_pair_entry->consume_q;
1230 			my_consume_q = queue_pair_entry->produce_q;
1231 			goto out;
1232 		}
1233 
1234 		result = VMCI_ERROR_ALREADY_EXISTS;
1235 		goto error_keep_entry;
1236 	}
1237 
1238 	my_produce_q = qp_alloc_queue(produce_size, flags);
1239 	if (!my_produce_q) {
1240 		pr_warn("Error allocating pages for produce queue\n");
1241 		result = VMCI_ERROR_NO_MEM;
1242 		goto error;
1243 	}
1244 
1245 	my_consume_q = qp_alloc_queue(consume_size, flags);
1246 	if (!my_consume_q) {
1247 		pr_warn("Error allocating pages for consume queue\n");
1248 		result = VMCI_ERROR_NO_MEM;
1249 		goto error;
1250 	}
1251 
1252 	queue_pair_entry = qp_guest_endpoint_create(*handle, peer, flags,
1253 						    produce_size, consume_size,
1254 						    my_produce_q, my_consume_q);
1255 	if (!queue_pair_entry) {
1256 		pr_warn("Error allocating memory in %s\n", __func__);
1257 		result = VMCI_ERROR_NO_MEM;
1258 		goto error;
1259 	}
1260 
1261 	result = qp_alloc_ppn_set(my_produce_q, num_produce_pages, my_consume_q,
1262 				  num_consume_pages,
1263 				  &queue_pair_entry->ppn_set);
1264 	if (result < VMCI_SUCCESS) {
1265 		pr_warn("qp_alloc_ppn_set failed\n");
1266 		goto error;
1267 	}
1268 
1269 	/*
1270 	 * It's only necessary to notify the host if this queue pair will be
1271 	 * attached to from another context.
1272 	 */
1273 	if (queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1274 		/* Local create case. */
1275 		u32 context_id = vmci_get_context_id();
1276 
1277 		/*
1278 		 * Enforce similar checks on local queue pairs as we
1279 		 * do for regular ones.  The handle's context must
1280 		 * match the creator or attacher context id (here they
1281 		 * are both the current context id) and the
1282 		 * attach-only flag cannot exist during create.  We
1283 		 * also ensure specified peer is this context or an
1284 		 * invalid one.
1285 		 */
1286 		if (queue_pair_entry->qp.handle.context != context_id ||
1287 		    (queue_pair_entry->qp.peer != VMCI_INVALID_ID &&
1288 		     queue_pair_entry->qp.peer != context_id)) {
1289 			result = VMCI_ERROR_NO_ACCESS;
1290 			goto error;
1291 		}
1292 
1293 		if (queue_pair_entry->qp.flags & VMCI_QPFLAG_ATTACH_ONLY) {
1294 			result = VMCI_ERROR_NOT_FOUND;
1295 			goto error;
1296 		}
1297 	} else {
1298 		result = qp_alloc_hypercall(queue_pair_entry);
1299 		if (result < VMCI_SUCCESS) {
1300 			pr_warn("qp_alloc_hypercall result = %d\n", result);
1301 			goto error;
1302 		}
1303 	}
1304 
1305 	qp_init_queue_mutex((struct vmci_queue *)my_produce_q,
1306 			    (struct vmci_queue *)my_consume_q);
1307 
1308 	qp_list_add_entry(&qp_guest_endpoints, &queue_pair_entry->qp);
1309 
1310  out:
1311 	queue_pair_entry->qp.ref_count++;
1312 	*handle = queue_pair_entry->qp.handle;
1313 	*produce_q = (struct vmci_queue *)my_produce_q;
1314 	*consume_q = (struct vmci_queue *)my_consume_q;
1315 
1316 	/*
1317 	 * We should initialize the queue pair header pages on a local
1318 	 * queue pair create.  For non-local queue pairs, the
1319 	 * hypervisor initializes the header pages in the create step.
1320 	 */
1321 	if ((queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) &&
1322 	    queue_pair_entry->qp.ref_count == 1) {
1323 		vmci_q_header_init((*produce_q)->q_header, *handle);
1324 		vmci_q_header_init((*consume_q)->q_header, *handle);
1325 	}
1326 
1327 	mutex_unlock(&qp_guest_endpoints.mutex);
1328 
1329 	return VMCI_SUCCESS;
1330 
1331  error:
1332 	mutex_unlock(&qp_guest_endpoints.mutex);
1333 	if (queue_pair_entry) {
1334 		/* The queues will be freed inside the destroy routine. */
1335 		qp_guest_endpoint_destroy(queue_pair_entry);
1336 	} else {
1337 		qp_free_queue(my_produce_q, produce_size);
1338 		qp_free_queue(my_consume_q, consume_size);
1339 	}
1340 	return result;
1341 
1342  error_keep_entry:
1343 	/* This path should only be used when an existing entry was found. */
1344 	mutex_unlock(&qp_guest_endpoints.mutex);
1345 	return result;
1346 }
1347 
1348 /*
1349  * The first endpoint issuing a queue pair allocation will create the state
1350  * of the queue pair in the queue pair broker.
1351  *
1352  * If the creator is a guest, it will associate a VMX virtual address range
1353  * with the queue pair as specified by the page_store. For compatibility with
1354  * older VMX'en, that would use a separate step to set the VMX virtual
1355  * address range, the virtual address range can be registered later using
1356  * vmci_qp_broker_set_page_store. In that case, a page_store of NULL should be
1357  * used.
1358  *
1359  * If the creator is the host, a page_store of NULL should be used as well,
1360  * since the host is not able to supply a page store for the queue pair.
1361  *
1362  * For older VMX and host callers, the queue pair will be created in the
1363  * VMCIQPB_CREATED_NO_MEM state, and for current VMX callers, it will be
1364  * created in VMCOQPB_CREATED_MEM state.
1365  */
1366 static int qp_broker_create(struct vmci_handle handle,
1367 			    u32 peer,
1368 			    u32 flags,
1369 			    u32 priv_flags,
1370 			    u64 produce_size,
1371 			    u64 consume_size,
1372 			    struct vmci_qp_page_store *page_store,
1373 			    struct vmci_ctx *context,
1374 			    vmci_event_release_cb wakeup_cb,
1375 			    void *client_data, struct qp_broker_entry **ent)
1376 {
1377 	struct qp_broker_entry *entry = NULL;
1378 	const u32 context_id = vmci_ctx_get_id(context);
1379 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1380 	int result;
1381 	u64 guest_produce_size;
1382 	u64 guest_consume_size;
1383 
1384 	/* Do not create if the caller asked not to. */
1385 	if (flags & VMCI_QPFLAG_ATTACH_ONLY)
1386 		return VMCI_ERROR_NOT_FOUND;
1387 
1388 	/*
1389 	 * Creator's context ID should match handle's context ID or the creator
1390 	 * must allow the context in handle's context ID as the "peer".
1391 	 */
1392 	if (handle.context != context_id && handle.context != peer)
1393 		return VMCI_ERROR_NO_ACCESS;
1394 
1395 	if (VMCI_CONTEXT_IS_VM(context_id) && VMCI_CONTEXT_IS_VM(peer))
1396 		return VMCI_ERROR_DST_UNREACHABLE;
1397 
1398 	/*
1399 	 * Creator's context ID for local queue pairs should match the
1400 	 * peer, if a peer is specified.
1401 	 */
1402 	if (is_local && peer != VMCI_INVALID_ID && context_id != peer)
1403 		return VMCI_ERROR_NO_ACCESS;
1404 
1405 	entry = kzalloc(sizeof(*entry), GFP_ATOMIC);
1406 	if (!entry)
1407 		return VMCI_ERROR_NO_MEM;
1408 
1409 	if (vmci_ctx_get_id(context) == VMCI_HOST_CONTEXT_ID && !is_local) {
1410 		/*
1411 		 * The queue pair broker entry stores values from the guest
1412 		 * point of view, so a creating host side endpoint should swap
1413 		 * produce and consume values -- unless it is a local queue
1414 		 * pair, in which case no swapping is necessary, since the local
1415 		 * attacher will swap queues.
1416 		 */
1417 
1418 		guest_produce_size = consume_size;
1419 		guest_consume_size = produce_size;
1420 	} else {
1421 		guest_produce_size = produce_size;
1422 		guest_consume_size = consume_size;
1423 	}
1424 
1425 	entry->qp.handle = handle;
1426 	entry->qp.peer = peer;
1427 	entry->qp.flags = flags;
1428 	entry->qp.produce_size = guest_produce_size;
1429 	entry->qp.consume_size = guest_consume_size;
1430 	entry->qp.ref_count = 1;
1431 	entry->create_id = context_id;
1432 	entry->attach_id = VMCI_INVALID_ID;
1433 	entry->state = VMCIQPB_NEW;
1434 	entry->require_trusted_attach =
1435 	    !!(context->priv_flags & VMCI_PRIVILEGE_FLAG_RESTRICTED);
1436 	entry->created_by_trusted =
1437 	    !!(priv_flags & VMCI_PRIVILEGE_FLAG_TRUSTED);
1438 	entry->vmci_page_files = false;
1439 	entry->wakeup_cb = wakeup_cb;
1440 	entry->client_data = client_data;
1441 	entry->produce_q = qp_host_alloc_queue(guest_produce_size);
1442 	if (entry->produce_q == NULL) {
1443 		result = VMCI_ERROR_NO_MEM;
1444 		goto error;
1445 	}
1446 	entry->consume_q = qp_host_alloc_queue(guest_consume_size);
1447 	if (entry->consume_q == NULL) {
1448 		result = VMCI_ERROR_NO_MEM;
1449 		goto error;
1450 	}
1451 
1452 	qp_init_queue_mutex(entry->produce_q, entry->consume_q);
1453 
1454 	INIT_LIST_HEAD(&entry->qp.list_item);
1455 
1456 	if (is_local) {
1457 		u8 *tmp;
1458 
1459 		entry->local_mem = kcalloc(QPE_NUM_PAGES(entry->qp),
1460 					   PAGE_SIZE, GFP_KERNEL);
1461 		if (entry->local_mem == NULL) {
1462 			result = VMCI_ERROR_NO_MEM;
1463 			goto error;
1464 		}
1465 		entry->state = VMCIQPB_CREATED_MEM;
1466 		entry->produce_q->q_header = entry->local_mem;
1467 		tmp = (u8 *)entry->local_mem + PAGE_SIZE *
1468 		    (DIV_ROUND_UP(entry->qp.produce_size, PAGE_SIZE) + 1);
1469 		entry->consume_q->q_header = (struct vmci_queue_header *)tmp;
1470 	} else if (page_store) {
1471 		/*
1472 		 * The VMX already initialized the queue pair headers, so no
1473 		 * need for the kernel side to do that.
1474 		 */
1475 		result = qp_host_register_user_memory(page_store,
1476 						      entry->produce_q,
1477 						      entry->consume_q);
1478 		if (result < VMCI_SUCCESS)
1479 			goto error;
1480 
1481 		entry->state = VMCIQPB_CREATED_MEM;
1482 	} else {
1483 		/*
1484 		 * A create without a page_store may be either a host
1485 		 * side create (in which case we are waiting for the
1486 		 * guest side to supply the memory) or an old style
1487 		 * queue pair create (in which case we will expect a
1488 		 * set page store call as the next step).
1489 		 */
1490 		entry->state = VMCIQPB_CREATED_NO_MEM;
1491 	}
1492 
1493 	qp_list_add_entry(&qp_broker_list, &entry->qp);
1494 	if (ent != NULL)
1495 		*ent = entry;
1496 
1497 	/* Add to resource obj */
1498 	result = vmci_resource_add(&entry->resource,
1499 				   VMCI_RESOURCE_TYPE_QPAIR_HOST,
1500 				   handle);
1501 	if (result != VMCI_SUCCESS) {
1502 		pr_warn("Failed to add new resource (handle=0x%x:0x%x), error: %d",
1503 			handle.context, handle.resource, result);
1504 		goto error;
1505 	}
1506 
1507 	entry->qp.handle = vmci_resource_handle(&entry->resource);
1508 	if (is_local) {
1509 		vmci_q_header_init(entry->produce_q->q_header,
1510 				   entry->qp.handle);
1511 		vmci_q_header_init(entry->consume_q->q_header,
1512 				   entry->qp.handle);
1513 	}
1514 
1515 	vmci_ctx_qp_create(context, entry->qp.handle);
1516 
1517 	return VMCI_SUCCESS;
1518 
1519  error:
1520 	if (entry != NULL) {
1521 		qp_host_free_queue(entry->produce_q, guest_produce_size);
1522 		qp_host_free_queue(entry->consume_q, guest_consume_size);
1523 		kfree(entry);
1524 	}
1525 
1526 	return result;
1527 }
1528 
1529 /*
1530  * Enqueues an event datagram to notify the peer VM attached to
1531  * the given queue pair handle about attach/detach event by the
1532  * given VM.  Returns Payload size of datagram enqueued on
1533  * success, error code otherwise.
1534  */
1535 static int qp_notify_peer(bool attach,
1536 			  struct vmci_handle handle,
1537 			  u32 my_id,
1538 			  u32 peer_id)
1539 {
1540 	int rv;
1541 	struct vmci_event_qp ev;
1542 
1543 	if (vmci_handle_is_invalid(handle) || my_id == VMCI_INVALID_ID ||
1544 	    peer_id == VMCI_INVALID_ID)
1545 		return VMCI_ERROR_INVALID_ARGS;
1546 
1547 	/*
1548 	 * In vmci_ctx_enqueue_datagram() we enforce the upper limit on
1549 	 * number of pending events from the hypervisor to a given VM
1550 	 * otherwise a rogue VM could do an arbitrary number of attach
1551 	 * and detach operations causing memory pressure in the host
1552 	 * kernel.
1553 	 */
1554 
1555 	ev.msg.hdr.dst = vmci_make_handle(peer_id, VMCI_EVENT_HANDLER);
1556 	ev.msg.hdr.src = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
1557 					  VMCI_CONTEXT_RESOURCE_ID);
1558 	ev.msg.hdr.payload_size = sizeof(ev) - sizeof(ev.msg.hdr);
1559 	ev.msg.event_data.event = attach ?
1560 	    VMCI_EVENT_QP_PEER_ATTACH : VMCI_EVENT_QP_PEER_DETACH;
1561 	ev.payload.handle = handle;
1562 	ev.payload.peer_id = my_id;
1563 
1564 	rv = vmci_datagram_dispatch(VMCI_HYPERVISOR_CONTEXT_ID,
1565 				    &ev.msg.hdr, false);
1566 	if (rv < VMCI_SUCCESS)
1567 		pr_warn("Failed to enqueue queue_pair %s event datagram for context (ID=0x%x)\n",
1568 			attach ? "ATTACH" : "DETACH", peer_id);
1569 
1570 	return rv;
1571 }
1572 
1573 /*
1574  * The second endpoint issuing a queue pair allocation will attach to
1575  * the queue pair registered with the queue pair broker.
1576  *
1577  * If the attacher is a guest, it will associate a VMX virtual address
1578  * range with the queue pair as specified by the page_store. At this
1579  * point, the already attach host endpoint may start using the queue
1580  * pair, and an attach event is sent to it. For compatibility with
1581  * older VMX'en, that used a separate step to set the VMX virtual
1582  * address range, the virtual address range can be registered later
1583  * using vmci_qp_broker_set_page_store. In that case, a page_store of
1584  * NULL should be used, and the attach event will be generated once
1585  * the actual page store has been set.
1586  *
1587  * If the attacher is the host, a page_store of NULL should be used as
1588  * well, since the page store information is already set by the guest.
1589  *
1590  * For new VMX and host callers, the queue pair will be moved to the
1591  * VMCIQPB_ATTACHED_MEM state, and for older VMX callers, it will be
1592  * moved to the VMCOQPB_ATTACHED_NO_MEM state.
1593  */
1594 static int qp_broker_attach(struct qp_broker_entry *entry,
1595 			    u32 peer,
1596 			    u32 flags,
1597 			    u32 priv_flags,
1598 			    u64 produce_size,
1599 			    u64 consume_size,
1600 			    struct vmci_qp_page_store *page_store,
1601 			    struct vmci_ctx *context,
1602 			    vmci_event_release_cb wakeup_cb,
1603 			    void *client_data,
1604 			    struct qp_broker_entry **ent)
1605 {
1606 	const u32 context_id = vmci_ctx_get_id(context);
1607 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1608 	int result;
1609 
1610 	if (entry->state != VMCIQPB_CREATED_NO_MEM &&
1611 	    entry->state != VMCIQPB_CREATED_MEM)
1612 		return VMCI_ERROR_UNAVAILABLE;
1613 
1614 	if (is_local) {
1615 		if (!(entry->qp.flags & VMCI_QPFLAG_LOCAL) ||
1616 		    context_id != entry->create_id) {
1617 			return VMCI_ERROR_INVALID_ARGS;
1618 		}
1619 	} else if (context_id == entry->create_id ||
1620 		   context_id == entry->attach_id) {
1621 		return VMCI_ERROR_ALREADY_EXISTS;
1622 	}
1623 
1624 	if (VMCI_CONTEXT_IS_VM(context_id) &&
1625 	    VMCI_CONTEXT_IS_VM(entry->create_id))
1626 		return VMCI_ERROR_DST_UNREACHABLE;
1627 
1628 	/*
1629 	 * If we are attaching from a restricted context then the queuepair
1630 	 * must have been created by a trusted endpoint.
1631 	 */
1632 	if ((context->priv_flags & VMCI_PRIVILEGE_FLAG_RESTRICTED) &&
1633 	    !entry->created_by_trusted)
1634 		return VMCI_ERROR_NO_ACCESS;
1635 
1636 	/*
1637 	 * If we are attaching to a queuepair that was created by a restricted
1638 	 * context then we must be trusted.
1639 	 */
1640 	if (entry->require_trusted_attach &&
1641 	    (!(priv_flags & VMCI_PRIVILEGE_FLAG_TRUSTED)))
1642 		return VMCI_ERROR_NO_ACCESS;
1643 
1644 	/*
1645 	 * If the creator specifies VMCI_INVALID_ID in "peer" field, access
1646 	 * control check is not performed.
1647 	 */
1648 	if (entry->qp.peer != VMCI_INVALID_ID && entry->qp.peer != context_id)
1649 		return VMCI_ERROR_NO_ACCESS;
1650 
1651 	if (entry->create_id == VMCI_HOST_CONTEXT_ID) {
1652 		/*
1653 		 * Do not attach if the caller doesn't support Host Queue Pairs
1654 		 * and a host created this queue pair.
1655 		 */
1656 
1657 		if (!vmci_ctx_supports_host_qp(context))
1658 			return VMCI_ERROR_INVALID_RESOURCE;
1659 
1660 	} else if (context_id == VMCI_HOST_CONTEXT_ID) {
1661 		struct vmci_ctx *create_context;
1662 		bool supports_host_qp;
1663 
1664 		/*
1665 		 * Do not attach a host to a user created queue pair if that
1666 		 * user doesn't support host queue pair end points.
1667 		 */
1668 
1669 		create_context = vmci_ctx_get(entry->create_id);
1670 		supports_host_qp = vmci_ctx_supports_host_qp(create_context);
1671 		vmci_ctx_put(create_context);
1672 
1673 		if (!supports_host_qp)
1674 			return VMCI_ERROR_INVALID_RESOURCE;
1675 	}
1676 
1677 	if ((entry->qp.flags & ~VMCI_QP_ASYMM) != (flags & ~VMCI_QP_ASYMM_PEER))
1678 		return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1679 
1680 	if (context_id != VMCI_HOST_CONTEXT_ID) {
1681 		/*
1682 		 * The queue pair broker entry stores values from the guest
1683 		 * point of view, so an attaching guest should match the values
1684 		 * stored in the entry.
1685 		 */
1686 
1687 		if (entry->qp.produce_size != produce_size ||
1688 		    entry->qp.consume_size != consume_size) {
1689 			return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1690 		}
1691 	} else if (entry->qp.produce_size != consume_size ||
1692 		   entry->qp.consume_size != produce_size) {
1693 		return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1694 	}
1695 
1696 	if (context_id != VMCI_HOST_CONTEXT_ID) {
1697 		/*
1698 		 * If a guest attached to a queue pair, it will supply
1699 		 * the backing memory.  If this is a pre NOVMVM vmx,
1700 		 * the backing memory will be supplied by calling
1701 		 * vmci_qp_broker_set_page_store() following the
1702 		 * return of the vmci_qp_broker_alloc() call. If it is
1703 		 * a vmx of version NOVMVM or later, the page store
1704 		 * must be supplied as part of the
1705 		 * vmci_qp_broker_alloc call.  Under all circumstances
1706 		 * must the initially created queue pair not have any
1707 		 * memory associated with it already.
1708 		 */
1709 
1710 		if (entry->state != VMCIQPB_CREATED_NO_MEM)
1711 			return VMCI_ERROR_INVALID_ARGS;
1712 
1713 		if (page_store != NULL) {
1714 			/*
1715 			 * Patch up host state to point to guest
1716 			 * supplied memory. The VMX already
1717 			 * initialized the queue pair headers, so no
1718 			 * need for the kernel side to do that.
1719 			 */
1720 
1721 			result = qp_host_register_user_memory(page_store,
1722 							      entry->produce_q,
1723 							      entry->consume_q);
1724 			if (result < VMCI_SUCCESS)
1725 				return result;
1726 
1727 			entry->state = VMCIQPB_ATTACHED_MEM;
1728 		} else {
1729 			entry->state = VMCIQPB_ATTACHED_NO_MEM;
1730 		}
1731 	} else if (entry->state == VMCIQPB_CREATED_NO_MEM) {
1732 		/*
1733 		 * The host side is attempting to attach to a queue
1734 		 * pair that doesn't have any memory associated with
1735 		 * it. This must be a pre NOVMVM vmx that hasn't set
1736 		 * the page store information yet, or a quiesced VM.
1737 		 */
1738 
1739 		return VMCI_ERROR_UNAVAILABLE;
1740 	} else {
1741 		/* The host side has successfully attached to a queue pair. */
1742 		entry->state = VMCIQPB_ATTACHED_MEM;
1743 	}
1744 
1745 	if (entry->state == VMCIQPB_ATTACHED_MEM) {
1746 		result =
1747 		    qp_notify_peer(true, entry->qp.handle, context_id,
1748 				   entry->create_id);
1749 		if (result < VMCI_SUCCESS)
1750 			pr_warn("Failed to notify peer (ID=0x%x) of attach to queue pair (handle=0x%x:0x%x)\n",
1751 				entry->create_id, entry->qp.handle.context,
1752 				entry->qp.handle.resource);
1753 	}
1754 
1755 	entry->attach_id = context_id;
1756 	entry->qp.ref_count++;
1757 	if (wakeup_cb) {
1758 		entry->wakeup_cb = wakeup_cb;
1759 		entry->client_data = client_data;
1760 	}
1761 
1762 	/*
1763 	 * When attaching to local queue pairs, the context already has
1764 	 * an entry tracking the queue pair, so don't add another one.
1765 	 */
1766 	if (!is_local)
1767 		vmci_ctx_qp_create(context, entry->qp.handle);
1768 
1769 	if (ent != NULL)
1770 		*ent = entry;
1771 
1772 	return VMCI_SUCCESS;
1773 }
1774 
1775 /*
1776  * queue_pair_Alloc for use when setting up queue pair endpoints
1777  * on the host.
1778  */
1779 static int qp_broker_alloc(struct vmci_handle handle,
1780 			   u32 peer,
1781 			   u32 flags,
1782 			   u32 priv_flags,
1783 			   u64 produce_size,
1784 			   u64 consume_size,
1785 			   struct vmci_qp_page_store *page_store,
1786 			   struct vmci_ctx *context,
1787 			   vmci_event_release_cb wakeup_cb,
1788 			   void *client_data,
1789 			   struct qp_broker_entry **ent,
1790 			   bool *swap)
1791 {
1792 	const u32 context_id = vmci_ctx_get_id(context);
1793 	bool create;
1794 	struct qp_broker_entry *entry = NULL;
1795 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1796 	int result;
1797 
1798 	if (vmci_handle_is_invalid(handle) ||
1799 	    (flags & ~VMCI_QP_ALL_FLAGS) || is_local ||
1800 	    !(produce_size || consume_size) ||
1801 	    !context || context_id == VMCI_INVALID_ID ||
1802 	    handle.context == VMCI_INVALID_ID) {
1803 		return VMCI_ERROR_INVALID_ARGS;
1804 	}
1805 
1806 	if (page_store && !VMCI_QP_PAGESTORE_IS_WELLFORMED(page_store))
1807 		return VMCI_ERROR_INVALID_ARGS;
1808 
1809 	/*
1810 	 * In the initial argument check, we ensure that non-vmkernel hosts
1811 	 * are not allowed to create local queue pairs.
1812 	 */
1813 
1814 	mutex_lock(&qp_broker_list.mutex);
1815 
1816 	if (!is_local && vmci_ctx_qp_exists(context, handle)) {
1817 		pr_devel("Context (ID=0x%x) already attached to queue pair (handle=0x%x:0x%x)\n",
1818 			 context_id, handle.context, handle.resource);
1819 		mutex_unlock(&qp_broker_list.mutex);
1820 		return VMCI_ERROR_ALREADY_EXISTS;
1821 	}
1822 
1823 	if (handle.resource != VMCI_INVALID_ID)
1824 		entry = qp_broker_handle_to_entry(handle);
1825 
1826 	if (!entry) {
1827 		create = true;
1828 		result =
1829 		    qp_broker_create(handle, peer, flags, priv_flags,
1830 				     produce_size, consume_size, page_store,
1831 				     context, wakeup_cb, client_data, ent);
1832 	} else {
1833 		create = false;
1834 		result =
1835 		    qp_broker_attach(entry, peer, flags, priv_flags,
1836 				     produce_size, consume_size, page_store,
1837 				     context, wakeup_cb, client_data, ent);
1838 	}
1839 
1840 	mutex_unlock(&qp_broker_list.mutex);
1841 
1842 	if (swap)
1843 		*swap = (context_id == VMCI_HOST_CONTEXT_ID) &&
1844 		    !(create && is_local);
1845 
1846 	return result;
1847 }
1848 
1849 /*
1850  * This function implements the kernel API for allocating a queue
1851  * pair.
1852  */
1853 static int qp_alloc_host_work(struct vmci_handle *handle,
1854 			      struct vmci_queue **produce_q,
1855 			      u64 produce_size,
1856 			      struct vmci_queue **consume_q,
1857 			      u64 consume_size,
1858 			      u32 peer,
1859 			      u32 flags,
1860 			      u32 priv_flags,
1861 			      vmci_event_release_cb wakeup_cb,
1862 			      void *client_data)
1863 {
1864 	struct vmci_handle new_handle;
1865 	struct vmci_ctx *context;
1866 	struct qp_broker_entry *entry;
1867 	int result;
1868 	bool swap;
1869 
1870 	if (vmci_handle_is_invalid(*handle)) {
1871 		new_handle = vmci_make_handle(
1872 			VMCI_HOST_CONTEXT_ID, VMCI_INVALID_ID);
1873 	} else
1874 		new_handle = *handle;
1875 
1876 	context = vmci_ctx_get(VMCI_HOST_CONTEXT_ID);
1877 	entry = NULL;
1878 	result =
1879 	    qp_broker_alloc(new_handle, peer, flags, priv_flags,
1880 			    produce_size, consume_size, NULL, context,
1881 			    wakeup_cb, client_data, &entry, &swap);
1882 	if (result == VMCI_SUCCESS) {
1883 		if (swap) {
1884 			/*
1885 			 * If this is a local queue pair, the attacher
1886 			 * will swap around produce and consume
1887 			 * queues.
1888 			 */
1889 
1890 			*produce_q = entry->consume_q;
1891 			*consume_q = entry->produce_q;
1892 		} else {
1893 			*produce_q = entry->produce_q;
1894 			*consume_q = entry->consume_q;
1895 		}
1896 
1897 		*handle = vmci_resource_handle(&entry->resource);
1898 	} else {
1899 		*handle = VMCI_INVALID_HANDLE;
1900 		pr_devel("queue pair broker failed to alloc (result=%d)\n",
1901 			 result);
1902 	}
1903 	vmci_ctx_put(context);
1904 	return result;
1905 }
1906 
1907 /*
1908  * Allocates a VMCI queue_pair. Only checks validity of input
1909  * arguments. The real work is done in the host or guest
1910  * specific function.
1911  */
1912 int vmci_qp_alloc(struct vmci_handle *handle,
1913 		  struct vmci_queue **produce_q,
1914 		  u64 produce_size,
1915 		  struct vmci_queue **consume_q,
1916 		  u64 consume_size,
1917 		  u32 peer,
1918 		  u32 flags,
1919 		  u32 priv_flags,
1920 		  bool guest_endpoint,
1921 		  vmci_event_release_cb wakeup_cb,
1922 		  void *client_data)
1923 {
1924 	if (!handle || !produce_q || !consume_q ||
1925 	    (!produce_size && !consume_size) || (flags & ~VMCI_QP_ALL_FLAGS))
1926 		return VMCI_ERROR_INVALID_ARGS;
1927 
1928 	if (guest_endpoint) {
1929 		return qp_alloc_guest_work(handle, produce_q,
1930 					   produce_size, consume_q,
1931 					   consume_size, peer,
1932 					   flags, priv_flags);
1933 	} else {
1934 		return qp_alloc_host_work(handle, produce_q,
1935 					  produce_size, consume_q,
1936 					  consume_size, peer, flags,
1937 					  priv_flags, wakeup_cb, client_data);
1938 	}
1939 }
1940 
1941 /*
1942  * This function implements the host kernel API for detaching from
1943  * a queue pair.
1944  */
1945 static int qp_detatch_host_work(struct vmci_handle handle)
1946 {
1947 	int result;
1948 	struct vmci_ctx *context;
1949 
1950 	context = vmci_ctx_get(VMCI_HOST_CONTEXT_ID);
1951 
1952 	result = vmci_qp_broker_detach(handle, context);
1953 
1954 	vmci_ctx_put(context);
1955 	return result;
1956 }
1957 
1958 /*
1959  * Detaches from a VMCI queue_pair. Only checks validity of input argument.
1960  * Real work is done in the host or guest specific function.
1961  */
1962 static int qp_detatch(struct vmci_handle handle, bool guest_endpoint)
1963 {
1964 	if (vmci_handle_is_invalid(handle))
1965 		return VMCI_ERROR_INVALID_ARGS;
1966 
1967 	if (guest_endpoint)
1968 		return qp_detatch_guest_work(handle);
1969 	else
1970 		return qp_detatch_host_work(handle);
1971 }
1972 
1973 /*
1974  * Returns the entry from the head of the list. Assumes that the list is
1975  * locked.
1976  */
1977 static struct qp_entry *qp_list_get_head(struct qp_list *qp_list)
1978 {
1979 	if (!list_empty(&qp_list->head)) {
1980 		struct qp_entry *entry =
1981 		    list_first_entry(&qp_list->head, struct qp_entry,
1982 				     list_item);
1983 		return entry;
1984 	}
1985 
1986 	return NULL;
1987 }
1988 
1989 void vmci_qp_broker_exit(void)
1990 {
1991 	struct qp_entry *entry;
1992 	struct qp_broker_entry *be;
1993 
1994 	mutex_lock(&qp_broker_list.mutex);
1995 
1996 	while ((entry = qp_list_get_head(&qp_broker_list))) {
1997 		be = (struct qp_broker_entry *)entry;
1998 
1999 		qp_list_remove_entry(&qp_broker_list, entry);
2000 		kfree(be);
2001 	}
2002 
2003 	mutex_unlock(&qp_broker_list.mutex);
2004 }
2005 
2006 /*
2007  * Requests that a queue pair be allocated with the VMCI queue
2008  * pair broker. Allocates a queue pair entry if one does not
2009  * exist. Attaches to one if it exists, and retrieves the page
2010  * files backing that queue_pair.  Assumes that the queue pair
2011  * broker lock is held.
2012  */
2013 int vmci_qp_broker_alloc(struct vmci_handle handle,
2014 			 u32 peer,
2015 			 u32 flags,
2016 			 u32 priv_flags,
2017 			 u64 produce_size,
2018 			 u64 consume_size,
2019 			 struct vmci_qp_page_store *page_store,
2020 			 struct vmci_ctx *context)
2021 {
2022 	return qp_broker_alloc(handle, peer, flags, priv_flags,
2023 			       produce_size, consume_size,
2024 			       page_store, context, NULL, NULL, NULL, NULL);
2025 }
2026 
2027 /*
2028  * VMX'en with versions lower than VMCI_VERSION_NOVMVM use a separate
2029  * step to add the UVAs of the VMX mapping of the queue pair. This function
2030  * provides backwards compatibility with such VMX'en, and takes care of
2031  * registering the page store for a queue pair previously allocated by the
2032  * VMX during create or attach. This function will move the queue pair state
2033  * to either from VMCIQBP_CREATED_NO_MEM to VMCIQBP_CREATED_MEM or
2034  * VMCIQBP_ATTACHED_NO_MEM to VMCIQBP_ATTACHED_MEM. If moving to the
2035  * attached state with memory, the queue pair is ready to be used by the
2036  * host peer, and an attached event will be generated.
2037  *
2038  * Assumes that the queue pair broker lock is held.
2039  *
2040  * This function is only used by the hosted platform, since there is no
2041  * issue with backwards compatibility for vmkernel.
2042  */
2043 int vmci_qp_broker_set_page_store(struct vmci_handle handle,
2044 				  u64 produce_uva,
2045 				  u64 consume_uva,
2046 				  struct vmci_ctx *context)
2047 {
2048 	struct qp_broker_entry *entry;
2049 	int result;
2050 	const u32 context_id = vmci_ctx_get_id(context);
2051 
2052 	if (vmci_handle_is_invalid(handle) || !context ||
2053 	    context_id == VMCI_INVALID_ID)
2054 		return VMCI_ERROR_INVALID_ARGS;
2055 
2056 	/*
2057 	 * We only support guest to host queue pairs, so the VMX must
2058 	 * supply UVAs for the mapped page files.
2059 	 */
2060 
2061 	if (produce_uva == 0 || consume_uva == 0)
2062 		return VMCI_ERROR_INVALID_ARGS;
2063 
2064 	mutex_lock(&qp_broker_list.mutex);
2065 
2066 	if (!vmci_ctx_qp_exists(context, handle)) {
2067 		pr_warn("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2068 			context_id, handle.context, handle.resource);
2069 		result = VMCI_ERROR_NOT_FOUND;
2070 		goto out;
2071 	}
2072 
2073 	entry = qp_broker_handle_to_entry(handle);
2074 	if (!entry) {
2075 		result = VMCI_ERROR_NOT_FOUND;
2076 		goto out;
2077 	}
2078 
2079 	/*
2080 	 * If I'm the owner then I can set the page store.
2081 	 *
2082 	 * Or, if a host created the queue_pair and I'm the attached peer
2083 	 * then I can set the page store.
2084 	 */
2085 	if (entry->create_id != context_id &&
2086 	    (entry->create_id != VMCI_HOST_CONTEXT_ID ||
2087 	     entry->attach_id != context_id)) {
2088 		result = VMCI_ERROR_QUEUEPAIR_NOTOWNER;
2089 		goto out;
2090 	}
2091 
2092 	if (entry->state != VMCIQPB_CREATED_NO_MEM &&
2093 	    entry->state != VMCIQPB_ATTACHED_NO_MEM) {
2094 		result = VMCI_ERROR_UNAVAILABLE;
2095 		goto out;
2096 	}
2097 
2098 	result = qp_host_get_user_memory(produce_uva, consume_uva,
2099 					 entry->produce_q, entry->consume_q);
2100 	if (result < VMCI_SUCCESS)
2101 		goto out;
2102 
2103 	result = qp_host_map_queues(entry->produce_q, entry->consume_q);
2104 	if (result < VMCI_SUCCESS) {
2105 		qp_host_unregister_user_memory(entry->produce_q,
2106 					       entry->consume_q);
2107 		goto out;
2108 	}
2109 
2110 	if (entry->state == VMCIQPB_CREATED_NO_MEM)
2111 		entry->state = VMCIQPB_CREATED_MEM;
2112 	else
2113 		entry->state = VMCIQPB_ATTACHED_MEM;
2114 
2115 	entry->vmci_page_files = true;
2116 
2117 	if (entry->state == VMCIQPB_ATTACHED_MEM) {
2118 		result =
2119 		    qp_notify_peer(true, handle, context_id, entry->create_id);
2120 		if (result < VMCI_SUCCESS) {
2121 			pr_warn("Failed to notify peer (ID=0x%x) of attach to queue pair (handle=0x%x:0x%x)\n",
2122 				entry->create_id, entry->qp.handle.context,
2123 				entry->qp.handle.resource);
2124 		}
2125 	}
2126 
2127 	result = VMCI_SUCCESS;
2128  out:
2129 	mutex_unlock(&qp_broker_list.mutex);
2130 	return result;
2131 }
2132 
2133 /*
2134  * Resets saved queue headers for the given QP broker
2135  * entry. Should be used when guest memory becomes available
2136  * again, or the guest detaches.
2137  */
2138 static void qp_reset_saved_headers(struct qp_broker_entry *entry)
2139 {
2140 	entry->produce_q->saved_header = NULL;
2141 	entry->consume_q->saved_header = NULL;
2142 }
2143 
2144 /*
2145  * The main entry point for detaching from a queue pair registered with the
2146  * queue pair broker. If more than one endpoint is attached to the queue
2147  * pair, the first endpoint will mainly decrement a reference count and
2148  * generate a notification to its peer. The last endpoint will clean up
2149  * the queue pair state registered with the broker.
2150  *
2151  * When a guest endpoint detaches, it will unmap and unregister the guest
2152  * memory backing the queue pair. If the host is still attached, it will
2153  * no longer be able to access the queue pair content.
2154  *
2155  * If the queue pair is already in a state where there is no memory
2156  * registered for the queue pair (any *_NO_MEM state), it will transition to
2157  * the VMCIQPB_SHUTDOWN_NO_MEM state. This will also happen, if a guest
2158  * endpoint is the first of two endpoints to detach. If the host endpoint is
2159  * the first out of two to detach, the queue pair will move to the
2160  * VMCIQPB_SHUTDOWN_MEM state.
2161  */
2162 int vmci_qp_broker_detach(struct vmci_handle handle, struct vmci_ctx *context)
2163 {
2164 	struct qp_broker_entry *entry;
2165 	const u32 context_id = vmci_ctx_get_id(context);
2166 	u32 peer_id;
2167 	bool is_local = false;
2168 	int result;
2169 
2170 	if (vmci_handle_is_invalid(handle) || !context ||
2171 	    context_id == VMCI_INVALID_ID) {
2172 		return VMCI_ERROR_INVALID_ARGS;
2173 	}
2174 
2175 	mutex_lock(&qp_broker_list.mutex);
2176 
2177 	if (!vmci_ctx_qp_exists(context, handle)) {
2178 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2179 			 context_id, handle.context, handle.resource);
2180 		result = VMCI_ERROR_NOT_FOUND;
2181 		goto out;
2182 	}
2183 
2184 	entry = qp_broker_handle_to_entry(handle);
2185 	if (!entry) {
2186 		pr_devel("Context (ID=0x%x) reports being attached to queue pair(handle=0x%x:0x%x) that isn't present in broker\n",
2187 			 context_id, handle.context, handle.resource);
2188 		result = VMCI_ERROR_NOT_FOUND;
2189 		goto out;
2190 	}
2191 
2192 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2193 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2194 		goto out;
2195 	}
2196 
2197 	if (context_id == entry->create_id) {
2198 		peer_id = entry->attach_id;
2199 		entry->create_id = VMCI_INVALID_ID;
2200 	} else {
2201 		peer_id = entry->create_id;
2202 		entry->attach_id = VMCI_INVALID_ID;
2203 	}
2204 	entry->qp.ref_count--;
2205 
2206 	is_local = entry->qp.flags & VMCI_QPFLAG_LOCAL;
2207 
2208 	if (context_id != VMCI_HOST_CONTEXT_ID) {
2209 		bool headers_mapped;
2210 
2211 		/*
2212 		 * Pre NOVMVM vmx'en may detach from a queue pair
2213 		 * before setting the page store, and in that case
2214 		 * there is no user memory to detach from. Also, more
2215 		 * recent VMX'en may detach from a queue pair in the
2216 		 * quiesced state.
2217 		 */
2218 
2219 		qp_acquire_queue_mutex(entry->produce_q);
2220 		headers_mapped = entry->produce_q->q_header ||
2221 		    entry->consume_q->q_header;
2222 		if (QPBROKERSTATE_HAS_MEM(entry)) {
2223 			result =
2224 			    qp_host_unmap_queues(INVALID_VMCI_GUEST_MEM_ID,
2225 						 entry->produce_q,
2226 						 entry->consume_q);
2227 			if (result < VMCI_SUCCESS)
2228 				pr_warn("Failed to unmap queue headers for queue pair (handle=0x%x:0x%x,result=%d)\n",
2229 					handle.context, handle.resource,
2230 					result);
2231 
2232 			if (entry->vmci_page_files)
2233 				qp_host_unregister_user_memory(entry->produce_q,
2234 							       entry->
2235 							       consume_q);
2236 			else
2237 				qp_host_unregister_user_memory(entry->produce_q,
2238 							       entry->
2239 							       consume_q);
2240 
2241 		}
2242 
2243 		if (!headers_mapped)
2244 			qp_reset_saved_headers(entry);
2245 
2246 		qp_release_queue_mutex(entry->produce_q);
2247 
2248 		if (!headers_mapped && entry->wakeup_cb)
2249 			entry->wakeup_cb(entry->client_data);
2250 
2251 	} else {
2252 		if (entry->wakeup_cb) {
2253 			entry->wakeup_cb = NULL;
2254 			entry->client_data = NULL;
2255 		}
2256 	}
2257 
2258 	if (entry->qp.ref_count == 0) {
2259 		qp_list_remove_entry(&qp_broker_list, &entry->qp);
2260 
2261 		if (is_local)
2262 			kfree(entry->local_mem);
2263 
2264 		qp_cleanup_queue_mutex(entry->produce_q, entry->consume_q);
2265 		qp_host_free_queue(entry->produce_q, entry->qp.produce_size);
2266 		qp_host_free_queue(entry->consume_q, entry->qp.consume_size);
2267 		/* Unlink from resource hash table and free callback */
2268 		vmci_resource_remove(&entry->resource);
2269 
2270 		kfree(entry);
2271 
2272 		vmci_ctx_qp_destroy(context, handle);
2273 	} else {
2274 		qp_notify_peer(false, handle, context_id, peer_id);
2275 		if (context_id == VMCI_HOST_CONTEXT_ID &&
2276 		    QPBROKERSTATE_HAS_MEM(entry)) {
2277 			entry->state = VMCIQPB_SHUTDOWN_MEM;
2278 		} else {
2279 			entry->state = VMCIQPB_SHUTDOWN_NO_MEM;
2280 		}
2281 
2282 		if (!is_local)
2283 			vmci_ctx_qp_destroy(context, handle);
2284 
2285 	}
2286 	result = VMCI_SUCCESS;
2287  out:
2288 	mutex_unlock(&qp_broker_list.mutex);
2289 	return result;
2290 }
2291 
2292 /*
2293  * Establishes the necessary mappings for a queue pair given a
2294  * reference to the queue pair guest memory. This is usually
2295  * called when a guest is unquiesced and the VMX is allowed to
2296  * map guest memory once again.
2297  */
2298 int vmci_qp_broker_map(struct vmci_handle handle,
2299 		       struct vmci_ctx *context,
2300 		       u64 guest_mem)
2301 {
2302 	struct qp_broker_entry *entry;
2303 	const u32 context_id = vmci_ctx_get_id(context);
2304 	bool is_local = false;
2305 	int result;
2306 
2307 	if (vmci_handle_is_invalid(handle) || !context ||
2308 	    context_id == VMCI_INVALID_ID)
2309 		return VMCI_ERROR_INVALID_ARGS;
2310 
2311 	mutex_lock(&qp_broker_list.mutex);
2312 
2313 	if (!vmci_ctx_qp_exists(context, handle)) {
2314 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2315 			 context_id, handle.context, handle.resource);
2316 		result = VMCI_ERROR_NOT_FOUND;
2317 		goto out;
2318 	}
2319 
2320 	entry = qp_broker_handle_to_entry(handle);
2321 	if (!entry) {
2322 		pr_devel("Context (ID=0x%x) reports being attached to queue pair (handle=0x%x:0x%x) that isn't present in broker\n",
2323 			 context_id, handle.context, handle.resource);
2324 		result = VMCI_ERROR_NOT_FOUND;
2325 		goto out;
2326 	}
2327 
2328 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2329 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2330 		goto out;
2331 	}
2332 
2333 	is_local = entry->qp.flags & VMCI_QPFLAG_LOCAL;
2334 	result = VMCI_SUCCESS;
2335 
2336 	if (context_id != VMCI_HOST_CONTEXT_ID) {
2337 		struct vmci_qp_page_store page_store;
2338 
2339 		page_store.pages = guest_mem;
2340 		page_store.len = QPE_NUM_PAGES(entry->qp);
2341 
2342 		qp_acquire_queue_mutex(entry->produce_q);
2343 		qp_reset_saved_headers(entry);
2344 		result =
2345 		    qp_host_register_user_memory(&page_store,
2346 						 entry->produce_q,
2347 						 entry->consume_q);
2348 		qp_release_queue_mutex(entry->produce_q);
2349 		if (result == VMCI_SUCCESS) {
2350 			/* Move state from *_NO_MEM to *_MEM */
2351 
2352 			entry->state++;
2353 
2354 			if (entry->wakeup_cb)
2355 				entry->wakeup_cb(entry->client_data);
2356 		}
2357 	}
2358 
2359  out:
2360 	mutex_unlock(&qp_broker_list.mutex);
2361 	return result;
2362 }
2363 
2364 /*
2365  * Saves a snapshot of the queue headers for the given QP broker
2366  * entry. Should be used when guest memory is unmapped.
2367  * Results:
2368  * VMCI_SUCCESS on success, appropriate error code if guest memory
2369  * can't be accessed..
2370  */
2371 static int qp_save_headers(struct qp_broker_entry *entry)
2372 {
2373 	int result;
2374 
2375 	if (entry->produce_q->saved_header != NULL &&
2376 	    entry->consume_q->saved_header != NULL) {
2377 		/*
2378 		 *  If the headers have already been saved, we don't need to do
2379 		 *  it again, and we don't want to map in the headers
2380 		 *  unnecessarily.
2381 		 */
2382 
2383 		return VMCI_SUCCESS;
2384 	}
2385 
2386 	if (NULL == entry->produce_q->q_header ||
2387 	    NULL == entry->consume_q->q_header) {
2388 		result = qp_host_map_queues(entry->produce_q, entry->consume_q);
2389 		if (result < VMCI_SUCCESS)
2390 			return result;
2391 	}
2392 
2393 	memcpy(&entry->saved_produce_q, entry->produce_q->q_header,
2394 	       sizeof(entry->saved_produce_q));
2395 	entry->produce_q->saved_header = &entry->saved_produce_q;
2396 	memcpy(&entry->saved_consume_q, entry->consume_q->q_header,
2397 	       sizeof(entry->saved_consume_q));
2398 	entry->consume_q->saved_header = &entry->saved_consume_q;
2399 
2400 	return VMCI_SUCCESS;
2401 }
2402 
2403 /*
2404  * Removes all references to the guest memory of a given queue pair, and
2405  * will move the queue pair from state *_MEM to *_NO_MEM. It is usually
2406  * called when a VM is being quiesced where access to guest memory should
2407  * avoided.
2408  */
2409 int vmci_qp_broker_unmap(struct vmci_handle handle,
2410 			 struct vmci_ctx *context,
2411 			 u32 gid)
2412 {
2413 	struct qp_broker_entry *entry;
2414 	const u32 context_id = vmci_ctx_get_id(context);
2415 	bool is_local = false;
2416 	int result;
2417 
2418 	if (vmci_handle_is_invalid(handle) || !context ||
2419 	    context_id == VMCI_INVALID_ID)
2420 		return VMCI_ERROR_INVALID_ARGS;
2421 
2422 	mutex_lock(&qp_broker_list.mutex);
2423 
2424 	if (!vmci_ctx_qp_exists(context, handle)) {
2425 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2426 			 context_id, handle.context, handle.resource);
2427 		result = VMCI_ERROR_NOT_FOUND;
2428 		goto out;
2429 	}
2430 
2431 	entry = qp_broker_handle_to_entry(handle);
2432 	if (!entry) {
2433 		pr_devel("Context (ID=0x%x) reports being attached to queue pair (handle=0x%x:0x%x) that isn't present in broker\n",
2434 			 context_id, handle.context, handle.resource);
2435 		result = VMCI_ERROR_NOT_FOUND;
2436 		goto out;
2437 	}
2438 
2439 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2440 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2441 		goto out;
2442 	}
2443 
2444 	is_local = entry->qp.flags & VMCI_QPFLAG_LOCAL;
2445 
2446 	if (context_id != VMCI_HOST_CONTEXT_ID) {
2447 		qp_acquire_queue_mutex(entry->produce_q);
2448 		result = qp_save_headers(entry);
2449 		if (result < VMCI_SUCCESS)
2450 			pr_warn("Failed to save queue headers for queue pair (handle=0x%x:0x%x,result=%d)\n",
2451 				handle.context, handle.resource, result);
2452 
2453 		qp_host_unmap_queues(gid, entry->produce_q, entry->consume_q);
2454 
2455 		/*
2456 		 * On hosted, when we unmap queue pairs, the VMX will also
2457 		 * unmap the guest memory, so we invalidate the previously
2458 		 * registered memory. If the queue pair is mapped again at a
2459 		 * later point in time, we will need to reregister the user
2460 		 * memory with a possibly new user VA.
2461 		 */
2462 		qp_host_unregister_user_memory(entry->produce_q,
2463 					       entry->consume_q);
2464 
2465 		/*
2466 		 * Move state from *_MEM to *_NO_MEM.
2467 		 */
2468 		entry->state--;
2469 
2470 		qp_release_queue_mutex(entry->produce_q);
2471 	}
2472 
2473 	result = VMCI_SUCCESS;
2474 
2475  out:
2476 	mutex_unlock(&qp_broker_list.mutex);
2477 	return result;
2478 }
2479 
2480 /*
2481  * Destroys all guest queue pair endpoints. If active guest queue
2482  * pairs still exist, hypercalls to attempt detach from these
2483  * queue pairs will be made. Any failure to detach is silently
2484  * ignored.
2485  */
2486 void vmci_qp_guest_endpoints_exit(void)
2487 {
2488 	struct qp_entry *entry;
2489 	struct qp_guest_endpoint *ep;
2490 
2491 	mutex_lock(&qp_guest_endpoints.mutex);
2492 
2493 	while ((entry = qp_list_get_head(&qp_guest_endpoints))) {
2494 		ep = (struct qp_guest_endpoint *)entry;
2495 
2496 		/* Don't make a hypercall for local queue_pairs. */
2497 		if (!(entry->flags & VMCI_QPFLAG_LOCAL))
2498 			qp_detatch_hypercall(entry->handle);
2499 
2500 		/* We cannot fail the exit, so let's reset ref_count. */
2501 		entry->ref_count = 0;
2502 		qp_list_remove_entry(&qp_guest_endpoints, entry);
2503 
2504 		qp_guest_endpoint_destroy(ep);
2505 	}
2506 
2507 	mutex_unlock(&qp_guest_endpoints.mutex);
2508 }
2509 
2510 /*
2511  * Helper routine that will lock the queue pair before subsequent
2512  * operations.
2513  * Note: Non-blocking on the host side is currently only implemented in ESX.
2514  * Since non-blocking isn't yet implemented on the host personality we
2515  * have no reason to acquire a spin lock.  So to avoid the use of an
2516  * unnecessary lock only acquire the mutex if we can block.
2517  */
2518 static void qp_lock(const struct vmci_qp *qpair)
2519 {
2520 	qp_acquire_queue_mutex(qpair->produce_q);
2521 }
2522 
2523 /*
2524  * Helper routine that unlocks the queue pair after calling
2525  * qp_lock.
2526  */
2527 static void qp_unlock(const struct vmci_qp *qpair)
2528 {
2529 	qp_release_queue_mutex(qpair->produce_q);
2530 }
2531 
2532 /*
2533  * The queue headers may not be mapped at all times. If a queue is
2534  * currently not mapped, it will be attempted to do so.
2535  */
2536 static int qp_map_queue_headers(struct vmci_queue *produce_q,
2537 				struct vmci_queue *consume_q)
2538 {
2539 	int result;
2540 
2541 	if (NULL == produce_q->q_header || NULL == consume_q->q_header) {
2542 		result = qp_host_map_queues(produce_q, consume_q);
2543 		if (result < VMCI_SUCCESS)
2544 			return (produce_q->saved_header &&
2545 				consume_q->saved_header) ?
2546 			    VMCI_ERROR_QUEUEPAIR_NOT_READY :
2547 			    VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2548 	}
2549 
2550 	return VMCI_SUCCESS;
2551 }
2552 
2553 /*
2554  * Helper routine that will retrieve the produce and consume
2555  * headers of a given queue pair. If the guest memory of the
2556  * queue pair is currently not available, the saved queue headers
2557  * will be returned, if these are available.
2558  */
2559 static int qp_get_queue_headers(const struct vmci_qp *qpair,
2560 				struct vmci_queue_header **produce_q_header,
2561 				struct vmci_queue_header **consume_q_header)
2562 {
2563 	int result;
2564 
2565 	result = qp_map_queue_headers(qpair->produce_q, qpair->consume_q);
2566 	if (result == VMCI_SUCCESS) {
2567 		*produce_q_header = qpair->produce_q->q_header;
2568 		*consume_q_header = qpair->consume_q->q_header;
2569 	} else if (qpair->produce_q->saved_header &&
2570 		   qpair->consume_q->saved_header) {
2571 		*produce_q_header = qpair->produce_q->saved_header;
2572 		*consume_q_header = qpair->consume_q->saved_header;
2573 		result = VMCI_SUCCESS;
2574 	}
2575 
2576 	return result;
2577 }
2578 
2579 /*
2580  * Callback from VMCI queue pair broker indicating that a queue
2581  * pair that was previously not ready, now either is ready or
2582  * gone forever.
2583  */
2584 static int qp_wakeup_cb(void *client_data)
2585 {
2586 	struct vmci_qp *qpair = (struct vmci_qp *)client_data;
2587 
2588 	qp_lock(qpair);
2589 	while (qpair->blocked > 0) {
2590 		qpair->blocked--;
2591 		qpair->generation++;
2592 		wake_up(&qpair->event);
2593 	}
2594 	qp_unlock(qpair);
2595 
2596 	return VMCI_SUCCESS;
2597 }
2598 
2599 /*
2600  * Makes the calling thread wait for the queue pair to become
2601  * ready for host side access.  Returns true when thread is
2602  * woken up after queue pair state change, false otherwise.
2603  */
2604 static bool qp_wait_for_ready_queue(struct vmci_qp *qpair)
2605 {
2606 	unsigned int generation;
2607 
2608 	qpair->blocked++;
2609 	generation = qpair->generation;
2610 	qp_unlock(qpair);
2611 	wait_event(qpair->event, generation != qpair->generation);
2612 	qp_lock(qpair);
2613 
2614 	return true;
2615 }
2616 
2617 /*
2618  * Enqueues a given buffer to the produce queue using the provided
2619  * function. As many bytes as possible (space available in the queue)
2620  * are enqueued.  Assumes the queue->mutex has been acquired.  Returns
2621  * VMCI_ERROR_QUEUEPAIR_NOSPACE if no space was available to enqueue
2622  * data, VMCI_ERROR_INVALID_SIZE, if any queue pointer is outside the
2623  * queue (as defined by the queue size), VMCI_ERROR_INVALID_ARGS, if
2624  * an error occured when accessing the buffer,
2625  * VMCI_ERROR_QUEUEPAIR_NOTATTACHED, if the queue pair pages aren't
2626  * available.  Otherwise, the number of bytes written to the queue is
2627  * returned.  Updates the tail pointer of the produce queue.
2628  */
2629 static ssize_t qp_enqueue_locked(struct vmci_queue *produce_q,
2630 				 struct vmci_queue *consume_q,
2631 				 const u64 produce_q_size,
2632 				 const void *buf,
2633 				 size_t buf_size,
2634 				 vmci_memcpy_to_queue_func memcpy_to_queue)
2635 {
2636 	s64 free_space;
2637 	u64 tail;
2638 	size_t written;
2639 	ssize_t result;
2640 
2641 	result = qp_map_queue_headers(produce_q, consume_q);
2642 	if (unlikely(result != VMCI_SUCCESS))
2643 		return result;
2644 
2645 	free_space = vmci_q_header_free_space(produce_q->q_header,
2646 					      consume_q->q_header,
2647 					      produce_q_size);
2648 	if (free_space == 0)
2649 		return VMCI_ERROR_QUEUEPAIR_NOSPACE;
2650 
2651 	if (free_space < VMCI_SUCCESS)
2652 		return (ssize_t) free_space;
2653 
2654 	written = (size_t) (free_space > buf_size ? buf_size : free_space);
2655 	tail = vmci_q_header_producer_tail(produce_q->q_header);
2656 	if (likely(tail + written < produce_q_size)) {
2657 		result = memcpy_to_queue(produce_q, tail, buf, 0, written);
2658 	} else {
2659 		/* Tail pointer wraps around. */
2660 
2661 		const size_t tmp = (size_t) (produce_q_size - tail);
2662 
2663 		result = memcpy_to_queue(produce_q, tail, buf, 0, tmp);
2664 		if (result >= VMCI_SUCCESS)
2665 			result = memcpy_to_queue(produce_q, 0, buf, tmp,
2666 						 written - tmp);
2667 	}
2668 
2669 	if (result < VMCI_SUCCESS)
2670 		return result;
2671 
2672 	vmci_q_header_add_producer_tail(produce_q->q_header, written,
2673 					produce_q_size);
2674 	return written;
2675 }
2676 
2677 /*
2678  * Dequeues data (if available) from the given consume queue. Writes data
2679  * to the user provided buffer using the provided function.
2680  * Assumes the queue->mutex has been acquired.
2681  * Results:
2682  * VMCI_ERROR_QUEUEPAIR_NODATA if no data was available to dequeue.
2683  * VMCI_ERROR_INVALID_SIZE, if any queue pointer is outside the queue
2684  * (as defined by the queue size).
2685  * VMCI_ERROR_INVALID_ARGS, if an error occured when accessing the buffer.
2686  * Otherwise the number of bytes dequeued is returned.
2687  * Side effects:
2688  * Updates the head pointer of the consume queue.
2689  */
2690 static ssize_t qp_dequeue_locked(struct vmci_queue *produce_q,
2691 				 struct vmci_queue *consume_q,
2692 				 const u64 consume_q_size,
2693 				 void *buf,
2694 				 size_t buf_size,
2695 				 vmci_memcpy_from_queue_func memcpy_from_queue,
2696 				 bool update_consumer)
2697 {
2698 	s64 buf_ready;
2699 	u64 head;
2700 	size_t read;
2701 	ssize_t result;
2702 
2703 	result = qp_map_queue_headers(produce_q, consume_q);
2704 	if (unlikely(result != VMCI_SUCCESS))
2705 		return result;
2706 
2707 	buf_ready = vmci_q_header_buf_ready(consume_q->q_header,
2708 					    produce_q->q_header,
2709 					    consume_q_size);
2710 	if (buf_ready == 0)
2711 		return VMCI_ERROR_QUEUEPAIR_NODATA;
2712 
2713 	if (buf_ready < VMCI_SUCCESS)
2714 		return (ssize_t) buf_ready;
2715 
2716 	read = (size_t) (buf_ready > buf_size ? buf_size : buf_ready);
2717 	head = vmci_q_header_consumer_head(produce_q->q_header);
2718 	if (likely(head + read < consume_q_size)) {
2719 		result = memcpy_from_queue(buf, 0, consume_q, head, read);
2720 	} else {
2721 		/* Head pointer wraps around. */
2722 
2723 		const size_t tmp = (size_t) (consume_q_size - head);
2724 
2725 		result = memcpy_from_queue(buf, 0, consume_q, head, tmp);
2726 		if (result >= VMCI_SUCCESS)
2727 			result = memcpy_from_queue(buf, tmp, consume_q, 0,
2728 						   read - tmp);
2729 
2730 	}
2731 
2732 	if (result < VMCI_SUCCESS)
2733 		return result;
2734 
2735 	if (update_consumer)
2736 		vmci_q_header_add_consumer_head(produce_q->q_header,
2737 						read, consume_q_size);
2738 
2739 	return read;
2740 }
2741 
2742 /*
2743  * vmci_qpair_alloc() - Allocates a queue pair.
2744  * @qpair:      Pointer for the new vmci_qp struct.
2745  * @handle:     Handle to track the resource.
2746  * @produce_qsize:      Desired size of the producer queue.
2747  * @consume_qsize:      Desired size of the consumer queue.
2748  * @peer:       ContextID of the peer.
2749  * @flags:      VMCI flags.
2750  * @priv_flags: VMCI priviledge flags.
2751  *
2752  * This is the client interface for allocating the memory for a
2753  * vmci_qp structure and then attaching to the underlying
2754  * queue.  If an error occurs allocating the memory for the
2755  * vmci_qp structure no attempt is made to attach.  If an
2756  * error occurs attaching, then the structure is freed.
2757  */
2758 int vmci_qpair_alloc(struct vmci_qp **qpair,
2759 		     struct vmci_handle *handle,
2760 		     u64 produce_qsize,
2761 		     u64 consume_qsize,
2762 		     u32 peer,
2763 		     u32 flags,
2764 		     u32 priv_flags)
2765 {
2766 	struct vmci_qp *my_qpair;
2767 	int retval;
2768 	struct vmci_handle src = VMCI_INVALID_HANDLE;
2769 	struct vmci_handle dst = vmci_make_handle(peer, VMCI_INVALID_ID);
2770 	enum vmci_route route;
2771 	vmci_event_release_cb wakeup_cb;
2772 	void *client_data;
2773 
2774 	/*
2775 	 * Restrict the size of a queuepair.  The device already
2776 	 * enforces a limit on the total amount of memory that can be
2777 	 * allocated to queuepairs for a guest.  However, we try to
2778 	 * allocate this memory before we make the queuepair
2779 	 * allocation hypercall.  On Linux, we allocate each page
2780 	 * separately, which means rather than fail, the guest will
2781 	 * thrash while it tries to allocate, and will become
2782 	 * increasingly unresponsive to the point where it appears to
2783 	 * be hung.  So we place a limit on the size of an individual
2784 	 * queuepair here, and leave the device to enforce the
2785 	 * restriction on total queuepair memory.  (Note that this
2786 	 * doesn't prevent all cases; a user with only this much
2787 	 * physical memory could still get into trouble.)  The error
2788 	 * used by the device is NO_RESOURCES, so use that here too.
2789 	 */
2790 
2791 	if (produce_qsize + consume_qsize < max(produce_qsize, consume_qsize) ||
2792 	    produce_qsize + consume_qsize > VMCI_MAX_GUEST_QP_MEMORY)
2793 		return VMCI_ERROR_NO_RESOURCES;
2794 
2795 	retval = vmci_route(&src, &dst, false, &route);
2796 	if (retval < VMCI_SUCCESS)
2797 		route = vmci_guest_code_active() ?
2798 		    VMCI_ROUTE_AS_GUEST : VMCI_ROUTE_AS_HOST;
2799 
2800 	if (flags & (VMCI_QPFLAG_NONBLOCK | VMCI_QPFLAG_PINNED)) {
2801 		pr_devel("NONBLOCK OR PINNED set");
2802 		return VMCI_ERROR_INVALID_ARGS;
2803 	}
2804 
2805 	my_qpair = kzalloc(sizeof(*my_qpair), GFP_KERNEL);
2806 	if (!my_qpair)
2807 		return VMCI_ERROR_NO_MEM;
2808 
2809 	my_qpair->produce_q_size = produce_qsize;
2810 	my_qpair->consume_q_size = consume_qsize;
2811 	my_qpair->peer = peer;
2812 	my_qpair->flags = flags;
2813 	my_qpair->priv_flags = priv_flags;
2814 
2815 	wakeup_cb = NULL;
2816 	client_data = NULL;
2817 
2818 	if (VMCI_ROUTE_AS_HOST == route) {
2819 		my_qpair->guest_endpoint = false;
2820 		if (!(flags & VMCI_QPFLAG_LOCAL)) {
2821 			my_qpair->blocked = 0;
2822 			my_qpair->generation = 0;
2823 			init_waitqueue_head(&my_qpair->event);
2824 			wakeup_cb = qp_wakeup_cb;
2825 			client_data = (void *)my_qpair;
2826 		}
2827 	} else {
2828 		my_qpair->guest_endpoint = true;
2829 	}
2830 
2831 	retval = vmci_qp_alloc(handle,
2832 			       &my_qpair->produce_q,
2833 			       my_qpair->produce_q_size,
2834 			       &my_qpair->consume_q,
2835 			       my_qpair->consume_q_size,
2836 			       my_qpair->peer,
2837 			       my_qpair->flags,
2838 			       my_qpair->priv_flags,
2839 			       my_qpair->guest_endpoint,
2840 			       wakeup_cb, client_data);
2841 
2842 	if (retval < VMCI_SUCCESS) {
2843 		kfree(my_qpair);
2844 		return retval;
2845 	}
2846 
2847 	*qpair = my_qpair;
2848 	my_qpair->handle = *handle;
2849 
2850 	return retval;
2851 }
2852 EXPORT_SYMBOL_GPL(vmci_qpair_alloc);
2853 
2854 /*
2855  * vmci_qpair_detach() - Detatches the client from a queue pair.
2856  * @qpair:      Reference of a pointer to the qpair struct.
2857  *
2858  * This is the client interface for detaching from a VMCIQPair.
2859  * Note that this routine will free the memory allocated for the
2860  * vmci_qp structure too.
2861  */
2862 int vmci_qpair_detach(struct vmci_qp **qpair)
2863 {
2864 	int result;
2865 	struct vmci_qp *old_qpair;
2866 
2867 	if (!qpair || !(*qpair))
2868 		return VMCI_ERROR_INVALID_ARGS;
2869 
2870 	old_qpair = *qpair;
2871 	result = qp_detatch(old_qpair->handle, old_qpair->guest_endpoint);
2872 
2873 	/*
2874 	 * The guest can fail to detach for a number of reasons, and
2875 	 * if it does so, it will cleanup the entry (if there is one).
2876 	 * The host can fail too, but it won't cleanup the entry
2877 	 * immediately, it will do that later when the context is
2878 	 * freed.  Either way, we need to release the qpair struct
2879 	 * here; there isn't much the caller can do, and we don't want
2880 	 * to leak.
2881 	 */
2882 
2883 	memset(old_qpair, 0, sizeof(*old_qpair));
2884 	old_qpair->handle = VMCI_INVALID_HANDLE;
2885 	old_qpair->peer = VMCI_INVALID_ID;
2886 	kfree(old_qpair);
2887 	*qpair = NULL;
2888 
2889 	return result;
2890 }
2891 EXPORT_SYMBOL_GPL(vmci_qpair_detach);
2892 
2893 /*
2894  * vmci_qpair_get_produce_indexes() - Retrieves the indexes of the producer.
2895  * @qpair:      Pointer to the queue pair struct.
2896  * @producer_tail:      Reference used for storing producer tail index.
2897  * @consumer_head:      Reference used for storing the consumer head index.
2898  *
2899  * This is the client interface for getting the current indexes of the
2900  * QPair from the point of the view of the caller as the producer.
2901  */
2902 int vmci_qpair_get_produce_indexes(const struct vmci_qp *qpair,
2903 				   u64 *producer_tail,
2904 				   u64 *consumer_head)
2905 {
2906 	struct vmci_queue_header *produce_q_header;
2907 	struct vmci_queue_header *consume_q_header;
2908 	int result;
2909 
2910 	if (!qpair)
2911 		return VMCI_ERROR_INVALID_ARGS;
2912 
2913 	qp_lock(qpair);
2914 	result =
2915 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2916 	if (result == VMCI_SUCCESS)
2917 		vmci_q_header_get_pointers(produce_q_header, consume_q_header,
2918 					   producer_tail, consumer_head);
2919 	qp_unlock(qpair);
2920 
2921 	if (result == VMCI_SUCCESS &&
2922 	    ((producer_tail && *producer_tail >= qpair->produce_q_size) ||
2923 	     (consumer_head && *consumer_head >= qpair->produce_q_size)))
2924 		return VMCI_ERROR_INVALID_SIZE;
2925 
2926 	return result;
2927 }
2928 EXPORT_SYMBOL_GPL(vmci_qpair_get_produce_indexes);
2929 
2930 /*
2931  * vmci_qpair_get_consume_indexes() - Retrieves the indexes of the comsumer.
2932  * @qpair:      Pointer to the queue pair struct.
2933  * @consumer_tail:      Reference used for storing consumer tail index.
2934  * @producer_head:      Reference used for storing the producer head index.
2935  *
2936  * This is the client interface for getting the current indexes of the
2937  * QPair from the point of the view of the caller as the consumer.
2938  */
2939 int vmci_qpair_get_consume_indexes(const struct vmci_qp *qpair,
2940 				   u64 *consumer_tail,
2941 				   u64 *producer_head)
2942 {
2943 	struct vmci_queue_header *produce_q_header;
2944 	struct vmci_queue_header *consume_q_header;
2945 	int result;
2946 
2947 	if (!qpair)
2948 		return VMCI_ERROR_INVALID_ARGS;
2949 
2950 	qp_lock(qpair);
2951 	result =
2952 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2953 	if (result == VMCI_SUCCESS)
2954 		vmci_q_header_get_pointers(consume_q_header, produce_q_header,
2955 					   consumer_tail, producer_head);
2956 	qp_unlock(qpair);
2957 
2958 	if (result == VMCI_SUCCESS &&
2959 	    ((consumer_tail && *consumer_tail >= qpair->consume_q_size) ||
2960 	     (producer_head && *producer_head >= qpair->consume_q_size)))
2961 		return VMCI_ERROR_INVALID_SIZE;
2962 
2963 	return result;
2964 }
2965 EXPORT_SYMBOL_GPL(vmci_qpair_get_consume_indexes);
2966 
2967 /*
2968  * vmci_qpair_produce_free_space() - Retrieves free space in producer queue.
2969  * @qpair:      Pointer to the queue pair struct.
2970  *
2971  * This is the client interface for getting the amount of free
2972  * space in the QPair from the point of the view of the caller as
2973  * the producer which is the common case.  Returns < 0 if err, else
2974  * available bytes into which data can be enqueued if > 0.
2975  */
2976 s64 vmci_qpair_produce_free_space(const struct vmci_qp *qpair)
2977 {
2978 	struct vmci_queue_header *produce_q_header;
2979 	struct vmci_queue_header *consume_q_header;
2980 	s64 result;
2981 
2982 	if (!qpair)
2983 		return VMCI_ERROR_INVALID_ARGS;
2984 
2985 	qp_lock(qpair);
2986 	result =
2987 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2988 	if (result == VMCI_SUCCESS)
2989 		result = vmci_q_header_free_space(produce_q_header,
2990 						  consume_q_header,
2991 						  qpair->produce_q_size);
2992 	else
2993 		result = 0;
2994 
2995 	qp_unlock(qpair);
2996 
2997 	return result;
2998 }
2999 EXPORT_SYMBOL_GPL(vmci_qpair_produce_free_space);
3000 
3001 /*
3002  * vmci_qpair_consume_free_space() - Retrieves free space in consumer queue.
3003  * @qpair:      Pointer to the queue pair struct.
3004  *
3005  * This is the client interface for getting the amount of free
3006  * space in the QPair from the point of the view of the caller as
3007  * the consumer which is not the common case.  Returns < 0 if err, else
3008  * available bytes into which data can be enqueued if > 0.
3009  */
3010 s64 vmci_qpair_consume_free_space(const struct vmci_qp *qpair)
3011 {
3012 	struct vmci_queue_header *produce_q_header;
3013 	struct vmci_queue_header *consume_q_header;
3014 	s64 result;
3015 
3016 	if (!qpair)
3017 		return VMCI_ERROR_INVALID_ARGS;
3018 
3019 	qp_lock(qpair);
3020 	result =
3021 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
3022 	if (result == VMCI_SUCCESS)
3023 		result = vmci_q_header_free_space(consume_q_header,
3024 						  produce_q_header,
3025 						  qpair->consume_q_size);
3026 	else
3027 		result = 0;
3028 
3029 	qp_unlock(qpair);
3030 
3031 	return result;
3032 }
3033 EXPORT_SYMBOL_GPL(vmci_qpair_consume_free_space);
3034 
3035 /*
3036  * vmci_qpair_produce_buf_ready() - Gets bytes ready to read from
3037  * producer queue.
3038  * @qpair:      Pointer to the queue pair struct.
3039  *
3040  * This is the client interface for getting the amount of
3041  * enqueued data in the QPair from the point of the view of the
3042  * caller as the producer which is not the common case.  Returns < 0 if err,
3043  * else available bytes that may be read.
3044  */
3045 s64 vmci_qpair_produce_buf_ready(const struct vmci_qp *qpair)
3046 {
3047 	struct vmci_queue_header *produce_q_header;
3048 	struct vmci_queue_header *consume_q_header;
3049 	s64 result;
3050 
3051 	if (!qpair)
3052 		return VMCI_ERROR_INVALID_ARGS;
3053 
3054 	qp_lock(qpair);
3055 	result =
3056 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
3057 	if (result == VMCI_SUCCESS)
3058 		result = vmci_q_header_buf_ready(produce_q_header,
3059 						 consume_q_header,
3060 						 qpair->produce_q_size);
3061 	else
3062 		result = 0;
3063 
3064 	qp_unlock(qpair);
3065 
3066 	return result;
3067 }
3068 EXPORT_SYMBOL_GPL(vmci_qpair_produce_buf_ready);
3069 
3070 /*
3071  * vmci_qpair_consume_buf_ready() - Gets bytes ready to read from
3072  * consumer queue.
3073  * @qpair:      Pointer to the queue pair struct.
3074  *
3075  * This is the client interface for getting the amount of
3076  * enqueued data in the QPair from the point of the view of the
3077  * caller as the consumer which is the normal case.  Returns < 0 if err,
3078  * else available bytes that may be read.
3079  */
3080 s64 vmci_qpair_consume_buf_ready(const struct vmci_qp *qpair)
3081 {
3082 	struct vmci_queue_header *produce_q_header;
3083 	struct vmci_queue_header *consume_q_header;
3084 	s64 result;
3085 
3086 	if (!qpair)
3087 		return VMCI_ERROR_INVALID_ARGS;
3088 
3089 	qp_lock(qpair);
3090 	result =
3091 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
3092 	if (result == VMCI_SUCCESS)
3093 		result = vmci_q_header_buf_ready(consume_q_header,
3094 						 produce_q_header,
3095 						 qpair->consume_q_size);
3096 	else
3097 		result = 0;
3098 
3099 	qp_unlock(qpair);
3100 
3101 	return result;
3102 }
3103 EXPORT_SYMBOL_GPL(vmci_qpair_consume_buf_ready);
3104 
3105 /*
3106  * vmci_qpair_enqueue() - Throw data on the queue.
3107  * @qpair:      Pointer to the queue pair struct.
3108  * @buf:        Pointer to buffer containing data
3109  * @buf_size:   Length of buffer.
3110  * @buf_type:   Buffer type (Unused).
3111  *
3112  * This is the client interface for enqueueing data into the queue.
3113  * Returns number of bytes enqueued or < 0 on error.
3114  */
3115 ssize_t vmci_qpair_enqueue(struct vmci_qp *qpair,
3116 			   const void *buf,
3117 			   size_t buf_size,
3118 			   int buf_type)
3119 {
3120 	ssize_t result;
3121 
3122 	if (!qpair || !buf)
3123 		return VMCI_ERROR_INVALID_ARGS;
3124 
3125 	qp_lock(qpair);
3126 
3127 	do {
3128 		result = qp_enqueue_locked(qpair->produce_q,
3129 					   qpair->consume_q,
3130 					   qpair->produce_q_size,
3131 					   buf, buf_size,
3132 					   qp_memcpy_to_queue);
3133 
3134 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3135 		    !qp_wait_for_ready_queue(qpair))
3136 			result = VMCI_ERROR_WOULD_BLOCK;
3137 
3138 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3139 
3140 	qp_unlock(qpair);
3141 
3142 	return result;
3143 }
3144 EXPORT_SYMBOL_GPL(vmci_qpair_enqueue);
3145 
3146 /*
3147  * vmci_qpair_dequeue() - Get data from the queue.
3148  * @qpair:      Pointer to the queue pair struct.
3149  * @buf:        Pointer to buffer for the data
3150  * @buf_size:   Length of buffer.
3151  * @buf_type:   Buffer type (Unused).
3152  *
3153  * This is the client interface for dequeueing data from the queue.
3154  * Returns number of bytes dequeued or < 0 on error.
3155  */
3156 ssize_t vmci_qpair_dequeue(struct vmci_qp *qpair,
3157 			   void *buf,
3158 			   size_t buf_size,
3159 			   int buf_type)
3160 {
3161 	ssize_t result;
3162 
3163 	if (!qpair || !buf)
3164 		return VMCI_ERROR_INVALID_ARGS;
3165 
3166 	qp_lock(qpair);
3167 
3168 	do {
3169 		result = qp_dequeue_locked(qpair->produce_q,
3170 					   qpair->consume_q,
3171 					   qpair->consume_q_size,
3172 					   buf, buf_size,
3173 					   qp_memcpy_from_queue, true);
3174 
3175 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3176 		    !qp_wait_for_ready_queue(qpair))
3177 			result = VMCI_ERROR_WOULD_BLOCK;
3178 
3179 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3180 
3181 	qp_unlock(qpair);
3182 
3183 	return result;
3184 }
3185 EXPORT_SYMBOL_GPL(vmci_qpair_dequeue);
3186 
3187 /*
3188  * vmci_qpair_peek() - Peek at the data in the queue.
3189  * @qpair:      Pointer to the queue pair struct.
3190  * @buf:        Pointer to buffer for the data
3191  * @buf_size:   Length of buffer.
3192  * @buf_type:   Buffer type (Unused on Linux).
3193  *
3194  * This is the client interface for peeking into a queue.  (I.e.,
3195  * copy data from the queue without updating the head pointer.)
3196  * Returns number of bytes dequeued or < 0 on error.
3197  */
3198 ssize_t vmci_qpair_peek(struct vmci_qp *qpair,
3199 			void *buf,
3200 			size_t buf_size,
3201 			int buf_type)
3202 {
3203 	ssize_t result;
3204 
3205 	if (!qpair || !buf)
3206 		return VMCI_ERROR_INVALID_ARGS;
3207 
3208 	qp_lock(qpair);
3209 
3210 	do {
3211 		result = qp_dequeue_locked(qpair->produce_q,
3212 					   qpair->consume_q,
3213 					   qpair->consume_q_size,
3214 					   buf, buf_size,
3215 					   qp_memcpy_from_queue, false);
3216 
3217 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3218 		    !qp_wait_for_ready_queue(qpair))
3219 			result = VMCI_ERROR_WOULD_BLOCK;
3220 
3221 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3222 
3223 	qp_unlock(qpair);
3224 
3225 	return result;
3226 }
3227 EXPORT_SYMBOL_GPL(vmci_qpair_peek);
3228 
3229 /*
3230  * vmci_qpair_enquev() - Throw data on the queue using iov.
3231  * @qpair:      Pointer to the queue pair struct.
3232  * @iov:        Pointer to buffer containing data
3233  * @iov_size:   Length of buffer.
3234  * @buf_type:   Buffer type (Unused).
3235  *
3236  * This is the client interface for enqueueing data into the queue.
3237  * This function uses IO vectors to handle the work. Returns number
3238  * of bytes enqueued or < 0 on error.
3239  */
3240 ssize_t vmci_qpair_enquev(struct vmci_qp *qpair,
3241 			  struct msghdr *msg,
3242 			  size_t iov_size,
3243 			  int buf_type)
3244 {
3245 	ssize_t result;
3246 
3247 	if (!qpair)
3248 		return VMCI_ERROR_INVALID_ARGS;
3249 
3250 	qp_lock(qpair);
3251 
3252 	do {
3253 		result = qp_enqueue_locked(qpair->produce_q,
3254 					   qpair->consume_q,
3255 					   qpair->produce_q_size,
3256 					   msg, iov_size,
3257 					   qp_memcpy_to_queue_iov);
3258 
3259 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3260 		    !qp_wait_for_ready_queue(qpair))
3261 			result = VMCI_ERROR_WOULD_BLOCK;
3262 
3263 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3264 
3265 	qp_unlock(qpair);
3266 
3267 	return result;
3268 }
3269 EXPORT_SYMBOL_GPL(vmci_qpair_enquev);
3270 
3271 /*
3272  * vmci_qpair_dequev() - Get data from the queue using iov.
3273  * @qpair:      Pointer to the queue pair struct.
3274  * @iov:        Pointer to buffer for the data
3275  * @iov_size:   Length of buffer.
3276  * @buf_type:   Buffer type (Unused).
3277  *
3278  * This is the client interface for dequeueing data from the queue.
3279  * This function uses IO vectors to handle the work. Returns number
3280  * of bytes dequeued or < 0 on error.
3281  */
3282 ssize_t vmci_qpair_dequev(struct vmci_qp *qpair,
3283 			  struct msghdr *msg,
3284 			  size_t iov_size,
3285 			  int buf_type)
3286 {
3287 	ssize_t result;
3288 
3289 	if (!qpair)
3290 		return VMCI_ERROR_INVALID_ARGS;
3291 
3292 	qp_lock(qpair);
3293 
3294 	do {
3295 		result = qp_dequeue_locked(qpair->produce_q,
3296 					   qpair->consume_q,
3297 					   qpair->consume_q_size,
3298 					   msg, iov_size,
3299 					   qp_memcpy_from_queue_iov,
3300 					   true);
3301 
3302 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3303 		    !qp_wait_for_ready_queue(qpair))
3304 			result = VMCI_ERROR_WOULD_BLOCK;
3305 
3306 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3307 
3308 	qp_unlock(qpair);
3309 
3310 	return result;
3311 }
3312 EXPORT_SYMBOL_GPL(vmci_qpair_dequev);
3313 
3314 /*
3315  * vmci_qpair_peekv() - Peek at the data in the queue using iov.
3316  * @qpair:      Pointer to the queue pair struct.
3317  * @iov:        Pointer to buffer for the data
3318  * @iov_size:   Length of buffer.
3319  * @buf_type:   Buffer type (Unused on Linux).
3320  *
3321  * This is the client interface for peeking into a queue.  (I.e.,
3322  * copy data from the queue without updating the head pointer.)
3323  * This function uses IO vectors to handle the work. Returns number
3324  * of bytes peeked or < 0 on error.
3325  */
3326 ssize_t vmci_qpair_peekv(struct vmci_qp *qpair,
3327 			 struct msghdr *msg,
3328 			 size_t iov_size,
3329 			 int buf_type)
3330 {
3331 	ssize_t result;
3332 
3333 	if (!qpair)
3334 		return VMCI_ERROR_INVALID_ARGS;
3335 
3336 	qp_lock(qpair);
3337 
3338 	do {
3339 		result = qp_dequeue_locked(qpair->produce_q,
3340 					   qpair->consume_q,
3341 					   qpair->consume_q_size,
3342 					   msg, iov_size,
3343 					   qp_memcpy_from_queue_iov,
3344 					   false);
3345 
3346 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3347 		    !qp_wait_for_ready_queue(qpair))
3348 			result = VMCI_ERROR_WOULD_BLOCK;
3349 
3350 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3351 
3352 	qp_unlock(qpair);
3353 	return result;
3354 }
3355 EXPORT_SYMBOL_GPL(vmci_qpair_peekv);
3356