#include <portals/types.h>
#include "ptlrpc_internal.h"
-extern int request_in_callback(ptl_event_t *ev);
+static LIST_HEAD (ptlrpc_all_services);
+static spinlock_t ptlrpc_all_services_lock = SPIN_LOCK_UNLOCKED;
-static int ptlrpc_check_event(struct ptlrpc_service *svc,
- struct ptlrpc_thread *thread, ptl_event_t *event)
+static void
+ptlrpc_free_server_req (struct ptlrpc_request *req)
{
- struct ptlrpc_srv_ni *srv_ni;
- int i, idx, rc;
- ENTRY;
+ /* The last request to be received into a request buffer uses space
+ * in the request buffer descriptor, otherwise requests are
+ * allocated dynamically in the incoming reply event handler */
+ if (req == &req->rq_rqbd->rqbd_req)
+ return;
- spin_lock(&svc->srv_lock);
+ OBD_FREE(req, sizeof(*req));
+}
+
+static char *
+ptlrpc_alloc_request_buffer (int size)
+{
+ char *ptr;
+
+ if (size > SVC_BUF_VMALLOC_THRESHOLD)
+ OBD_VMALLOC(ptr, size);
+ else
+ OBD_ALLOC(ptr, size);
+
+ return (ptr);
+}
- if (thread->t_flags & SVC_STOPPING)
- GOTO(out, rc = 1);
+static void
+ptlrpc_free_request_buffer (char *ptr, int size)
+{
+ if (size > SVC_BUF_VMALLOC_THRESHOLD)
+ OBD_VFREE(ptr, size);
+ else
+ OBD_FREE(ptr, size);
+}
- LASSERT ((thread->t_flags & SVC_EVENT) == 0);
- LASSERT (ptlrpc_ninterfaces > 0);
+struct ptlrpc_request_buffer_desc *
+ptlrpc_alloc_rqbd (struct ptlrpc_srv_ni *srv_ni)
+{
+ struct ptlrpc_service *svc = srv_ni->sni_service;
+ unsigned long flags;
+ struct ptlrpc_request_buffer_desc *rqbd;
+
+ OBD_ALLOC(rqbd, sizeof (*rqbd));
+ if (rqbd == NULL)
+ return (NULL);
+
+ rqbd->rqbd_srv_ni = srv_ni;
+ rqbd->rqbd_refcount = 0;
+ rqbd->rqbd_cbid.cbid_fn = request_in_callback;
+ rqbd->rqbd_cbid.cbid_arg = rqbd;
+ rqbd->rqbd_buffer = ptlrpc_alloc_request_buffer(svc->srv_buf_size);
+
+ if (rqbd->rqbd_buffer == NULL) {
+ OBD_FREE(rqbd, sizeof (*rqbd));
+ return (NULL);
+ }
- for (i = 0; i < ptlrpc_ninterfaces; i++) {
- idx = (svc->srv_interface_rover + i) % ptlrpc_ninterfaces;
- srv_ni = &svc->srv_interfaces[idx];
+ spin_lock_irqsave (&svc->srv_lock, flags);
+ list_add(&rqbd->rqbd_list, &srv_ni->sni_rqbds);
+ svc->srv_nbufs++;
+ spin_unlock_irqrestore (&svc->srv_lock, flags);
- LASSERT (!PtlHandleEqual (srv_ni->sni_eq_h, PTL_HANDLE_NONE));
+ return (rqbd);
+}
- rc = PtlEQGet(srv_ni->sni_eq_h, event);
- switch (rc) {
- case PTL_OK:
- /* next time start with the next interface */
- svc->srv_interface_rover = (idx+1) % ptlrpc_ninterfaces;
- thread->t_flags |= SVC_EVENT;
- GOTO(out, rc = 1);
+void
+ptlrpc_free_rqbd (struct ptlrpc_request_buffer_desc *rqbd)
+{
+ struct ptlrpc_srv_ni *sni = rqbd->rqbd_srv_ni;
+ struct ptlrpc_service *svc = sni->sni_service;
+ unsigned long flags;
+
+ LASSERT (rqbd->rqbd_refcount == 0);
+
+ spin_lock_irqsave(&svc->srv_lock, flags);
+ list_del(&rqbd->rqbd_list);
+ svc->srv_nbufs--;
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ ptlrpc_free_request_buffer (rqbd->rqbd_buffer, svc->srv_buf_size);
+ OBD_FREE (rqbd, sizeof (*rqbd));
+}
- case PTL_EQ_EMPTY:
- continue;
+void
+ptlrpc_save_lock (struct ptlrpc_request *req,
+ struct lustre_handle *lock, int mode)
+{
+ struct ptlrpc_reply_state *rs = req->rq_reply_state;
+ int idx;
- case PTL_EQ_DROPPED:
- CWARN("Event queue overflow (bug 2125): timeouts will "
- "follow.\n");
- continue;
+ LASSERT (rs != NULL);
+ LASSERT (rs->rs_nlocks < RS_MAX_LOCKS);
+
+ idx = rs->rs_nlocks++;
+ rs->rs_locks[idx] = *lock;
+ rs->rs_modes[idx] = mode;
+ rs->rs_difficult = 1;
+}
+
+void
+ptlrpc_schedule_difficult_reply (struct ptlrpc_reply_state *rs)
+{
+ struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
+
+#ifdef CONFIG_SMP
+ LASSERT (spin_is_locked (&svc->srv_lock));
+#endif
+ LASSERT (rs->rs_difficult);
+ rs->rs_scheduled_ever = 1; /* flag any notification attempt */
+
+ if (rs->rs_scheduled) /* being set up or already notified */
+ return;
+
+ rs->rs_scheduled = 1;
+ list_del (&rs->rs_list);
+ list_add (&rs->rs_list, &svc->srv_reply_queue);
+ wake_up (&svc->srv_waitq);
+}
+
+void
+ptlrpc_commit_replies (struct obd_device *obd)
+{
+ struct list_head *tmp;
+ struct list_head *nxt;
+ unsigned long flags;
+
+ /* Find any replies that have been committed and get their service
+ * to attend to complete them. */
+
+ /* CAVEAT EMPTOR: spinlock ordering!!! */
+ spin_lock_irqsave (&obd->obd_uncommitted_replies_lock, flags);
+
+ list_for_each_safe (tmp, nxt, &obd->obd_uncommitted_replies) {
+ struct ptlrpc_reply_state *rs =
+ list_entry (tmp, struct ptlrpc_reply_state, rs_obd_list);
+
+ LASSERT (rs->rs_difficult);
- default:
- CERROR("BUG: PtlEQGet returned %d\n", rc);
- LBUG();
+ if (rs->rs_transno <= obd->obd_last_committed) {
+ struct ptlrpc_service *svc = rs->rs_srv_ni->sni_service;
+
+ spin_lock (&svc->srv_lock);
+ list_del_init (&rs->rs_obd_list);
+ ptlrpc_schedule_difficult_reply (rs);
+ spin_unlock (&svc->srv_lock);
}
}
- rc = 0;
- EXIT;
- out:
- spin_unlock(&svc->srv_lock);
- return rc;
+
+ spin_unlock_irqrestore (&obd->obd_uncommitted_replies_lock, flags);
+}
+
+static long
+timeval_sub(struct timeval *large, struct timeval *small)
+{
+ return (large->tv_sec - small->tv_sec) * 1000000 +
+ (large->tv_usec - small->tv_usec);
}
-struct ptlrpc_service * ptlrpc_init_svc(__u32 nevents, __u32 nbufs,
- __u32 bufsize, __u32 max_req_size,
- int req_portal, int rep_portal,
- svc_handler_t handler, char *name,
- struct proc_dir_entry *proc_entry)
+struct ptlrpc_service *
+ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size,
+ int req_portal, int rep_portal,
+ svc_handler_t handler, char *name,
+ struct proc_dir_entry *proc_entry)
{
- int i, j, ssize, rc;
- struct ptlrpc_service *service;
- struct ptlrpc_srv_ni *srv_ni;
+ int i;
+ int j;
+ int ssize;
+ struct ptlrpc_service *service;
+ struct ptlrpc_srv_ni *srv_ni;
+ struct ptlrpc_request_buffer_desc *rqbd;
ENTRY;
LASSERT (ptlrpc_ninterfaces > 0);
-
+ LASSERT (nbufs > 0);
+ LASSERT (bufsize >= max_req_size);
+
ssize = offsetof (struct ptlrpc_service,
srv_interfaces[ptlrpc_ninterfaces]);
OBD_ALLOC(service, ssize);
service->srv_max_req_size = max_req_size;
service->srv_buf_size = bufsize;
-
service->srv_rep_portal = rep_portal;
service->srv_req_portal = req_portal;
service->srv_handler = handler;
- service->srv_interface_rover = 0;
+
+ INIT_LIST_HEAD(&service->srv_request_queue);
+ INIT_LIST_HEAD(&service->srv_reply_queue);
/* First initialise enough for early teardown */
for (i = 0; i < ptlrpc_ninterfaces; i++) {
srv_ni->sni_service = service;
srv_ni->sni_ni = &ptlrpc_interfaces[i];
- srv_ni->sni_eq_h = PTL_HANDLE_NONE;
INIT_LIST_HEAD(&srv_ni->sni_rqbds);
- srv_ni->sni_nrqbds = 0;
- atomic_set(&srv_ni->sni_nrqbds_receiving, 0);
+ INIT_LIST_HEAD(&srv_ni->sni_active_replies);
}
- /* Now allocate the event queue and request buffers, assuming all
- * interfaces require the same level of buffering. */
+ spin_lock (&ptlrpc_all_services_lock);
+ list_add (&service->srv_list, &ptlrpc_all_services);
+ spin_unlock (&ptlrpc_all_services_lock);
+
+ /* Now allocate the request buffers, assuming all interfaces require
+ * the same number. */
for (i = 0; i < ptlrpc_ninterfaces; i++) {
srv_ni = &service->srv_interfaces[i];
CDEBUG (D_NET, "%s: initialising interface %s\n", name,
srv_ni->sni_ni->pni_name);
- rc = PtlEQAlloc(srv_ni->sni_ni->pni_ni_h, nevents,
- request_in_callback, &(srv_ni->sni_eq_h));
- if (rc != PTL_OK) {
- CERROR("%s.%d: PtlEQAlloc on %s failed: %d\n",
- name, i, srv_ni->sni_ni->pni_name, rc);
- GOTO (failed, NULL);
- }
-
for (j = 0; j < nbufs; j++) {
- struct ptlrpc_request_buffer_desc *rqbd;
-
- OBD_ALLOC_WAIT(rqbd, sizeof(*rqbd));
+ rqbd = ptlrpc_alloc_rqbd (srv_ni);
+
if (rqbd == NULL) {
- CERROR ("%s.%d: Can't allocate request "
- "descriptor %d on %s\n",
- name, i, srv_ni->sni_nrqbds,
+ CERROR ("%s.%d: Can't allocate request %d "
+ "on %s\n", name, i, j,
srv_ni->sni_ni->pni_name);
GOTO(failed, NULL);
}
-
- rqbd->rqbd_srv_ni = srv_ni;
- rqbd->rqbd_me_h = PTL_HANDLE_NONE;
- atomic_set(&rqbd->rqbd_refcount, 0);
-
- OBD_ALLOC_WAIT(rqbd->rqbd_buffer, service->srv_buf_size);
- if (rqbd->rqbd_buffer == NULL) {
- CERROR ("%s.%d: Can't allocate request "
- "buffer %d on %s\n",
- name, i, srv_ni->sni_nrqbds,
- srv_ni->sni_ni->pni_name);
- OBD_FREE(rqbd, sizeof(*rqbd));
- GOTO(failed, NULL);
- }
- list_add(&rqbd->rqbd_list, &srv_ni->sni_rqbds);
- srv_ni->sni_nrqbds++;
-
- ptlrpc_link_svc_me(rqbd);
+ ptlrpc_register_rqbd (rqbd);
}
}
return NULL;
}
-static int handle_incoming_request(struct obd_device *obddev,
- struct ptlrpc_service *svc,
- ptl_event_t *event,
- struct ptlrpc_request *request)
+static int
+ptlrpc_server_handle_request (struct ptlrpc_service *svc)
{
- struct ptlrpc_request_buffer_desc *rqbd = event->mem_desc.user_ptr;
- int rc;
-
- /* FIXME: If we move to an event-driven model, we should put the request
- * on the stack of mds_handle instead. */
+ struct ptlrpc_request *request;
+ unsigned long flags;
+ struct timeval work_start;
+ struct timeval work_end;
+ long timediff;
+ int refcount;
+ int rc;
+ ENTRY;
- LASSERT (atomic_read (&rqbd->rqbd_refcount) > 0);
- LASSERT ((event->mem_desc.options & (PTL_MD_IOV | PTL_MD_KIOV)) == 0);
- LASSERT (rqbd->rqbd_srv_ni->sni_service == svc);
- LASSERT (rqbd->rqbd_buffer == event->mem_desc.start);
- LASSERT (event->offset + event->mlength <= svc->srv_buf_size);
+ spin_lock_irqsave (&svc->srv_lock, flags);
+ if (list_empty (&svc->srv_request_queue) ||
+ (svc->srv_n_difficult_replies != 0 &&
+ svc->srv_n_active_reqs >= (svc->srv_nthreads - 1))) {
+ /* If all the other threads are handling requests, I must
+ * remain free to handle any 'difficult' reply that might
+ * block them */
+ spin_unlock_irqrestore (&svc->srv_lock, flags);
+ RETURN(0);
+ }
- memset(request, 0, sizeof(*request));
- spin_lock_init (&request->rq_lock);
- INIT_LIST_HEAD(&request->rq_list);
- request->rq_svc = svc;
- request->rq_xid = event->match_bits;
- request->rq_reqmsg = event->mem_desc.start + event->offset;
- request->rq_reqlen = event->mlength;
+ request = list_entry (svc->srv_request_queue.next,
+ struct ptlrpc_request, rq_list);
+ list_del_init (&request->rq_list);
+ svc->srv_n_queued_reqs--;
+ svc->srv_n_active_reqs++;
+
+ spin_unlock_irqrestore (&svc->srv_lock, flags);
+
+ do_gettimeofday(&work_start);
+ timediff = timeval_sub(&work_start, &request->rq_arrival_time);
+ if (svc->srv_stats != NULL) {
+ lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
+ timediff);
+ lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
+ svc->srv_n_queued_reqs);
+ lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
+ svc->srv_n_active_reqs);
+ }
#if SWAB_PARANOIA
/* Clear request swab mask; this is a new request */
if (rc != 0) {
CERROR ("error unpacking request: ptl %d from "LPX64
" xid "LPU64"\n", svc->srv_req_portal,
- event->initiator.nid, request->rq_xid);
+ request->rq_peer.peer_nid, request->rq_xid);
goto out;
}
+
rc = -EINVAL;
if (request->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
- CERROR("wrong packet type received (type=%u)\n",
- request->rq_reqmsg->type);
+ CERROR("wrong packet type received (type=%u) from "
+ LPX64"\n", request->rq_reqmsg->type,
+ request->rq_peer.peer_nid);
goto out;
}
- CDEBUG(D_NET, "got req "LPD64" (md: %p + %d)\n", request->rq_xid,
- event->mem_desc.start, event->offset);
+ CDEBUG(D_NET, "got req "LPD64"\n", request->rq_xid);
- request->rq_peer.peer_nid = event->initiator.nid;
- request->rq_peer.peer_ni = rqbd->rqbd_srv_ni->sni_ni;
+ /* Discard requests queued for longer than my timeout. If the
+ * client's timeout is similar to mine, she'll be timing out this
+ * REQ anyway (bug 1502) */
+ if (timediff / 1000000 > (long)obd_timeout) {
+ CERROR("Dropping timed-out request from "LPX64
+ ": %ld seconds old\n",
+ request->rq_peer.peer_nid, timediff / 1000000);
+ goto out;
+ }
request->rq_export = class_conn2export(&request->rq_reqmsg->handle);
if (request->rq_export) {
- if (request->rq_reqmsg->conn_cnt <
+ if (request->rq_reqmsg->conn_cnt <
request->rq_export->exp_conn_cnt) {
DEBUG_REQ(D_ERROR, request,
"DROPPING req from old connection %d < %d",
request->rq_export->exp_last_request_time =
LTIME_S(CURRENT_TIME);
- }
+ }
CDEBUG(D_RPCTRACE, "Handling RPC pname:cluuid+ref:pid:xid:ni:nid:opc "
"%s:%s+%d:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
(request->rq_export ?
atomic_read(&request->rq_export->exp_refcount) : -99),
request->rq_reqmsg->status, request->rq_xid,
- rqbd->rqbd_srv_ni->sni_ni->pni_name, event->initiator.nid,
+ request->rq_peer.peer_ni->pni_name,
+ request->rq_peer.peer_nid,
request->rq_reqmsg->opc);
rc = svc->srv_handler(request);
(request->rq_export ?
atomic_read(&request->rq_export->exp_refcount) : -99),
request->rq_reqmsg->status, request->rq_xid,
- rqbd->rqbd_srv_ni->sni_ni->pni_name, event->initiator.nid,
+ request->rq_peer.peer_ni->pni_name,
+ request->rq_peer.peer_nid,
request->rq_reqmsg->opc);
put_conn:
class_export_put(request->rq_export);
out:
- if (atomic_dec_and_test (&rqbd->rqbd_refcount)) /* last reference? */
- ptlrpc_link_svc_me (rqbd);
+ do_gettimeofday(&work_end);
+
+ timediff = timeval_sub(&work_end, &work_start);
+
+ CDEBUG((timediff / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
+ "request "LPU64" opc %u from NID "LPX64" processed in %ldus "
+ "(%ldus total)\n", request->rq_xid, request->rq_reqmsg->opc,
+ request->rq_peer.peer_nid,
+ timediff, timeval_sub(&work_end, &request->rq_arrival_time));
+
+ if (svc->srv_stats != NULL) {
+ int opc = opcode_offset(request->rq_reqmsg->opc);
+ if (opc > 0) {
+ LASSERT(opc < LUSTRE_MAX_OPCODES);
+ lprocfs_counter_add(svc->srv_stats,
+ opc + PTLRPC_LAST_CNTR,
+ timediff);
+ }
+ }
+
+ spin_lock_irqsave(&svc->srv_lock, flags);
+ svc->srv_n_active_reqs--;
+ refcount = --(request->rq_rqbd->rqbd_refcount);
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ if (refcount == 0) {
+ /* rqbd now idle: repost */
+ ptlrpc_register_rqbd(request->rq_rqbd);
+ }
+
+ ptlrpc_free_server_req(request);
+
+ RETURN(1);
+}
+
+static int
+ptlrpc_server_handle_reply (struct ptlrpc_service *svc)
+{
+ struct ptlrpc_reply_state *rs;
+ unsigned long flags;
+ struct obd_export *exp;
+ struct obd_device *obd;
+ int nlocks;
+ int been_handled;
+ ENTRY;
+
+ spin_lock_irqsave (&svc->srv_lock, flags);
+ if (list_empty (&svc->srv_reply_queue)) {
+ spin_unlock_irqrestore (&svc->srv_lock, flags);
+ RETURN(0);
+ }
+
+ rs = list_entry (svc->srv_reply_queue.next,
+ struct ptlrpc_reply_state, rs_list);
+
+ exp = rs->rs_export;
+ obd = exp->exp_obd;
+
+ LASSERT (rs->rs_difficult);
+ LASSERT (rs->rs_scheduled);
+
+ list_del_init (&rs->rs_list);
+
+ /* Disengage from notifiers carefully (lock ordering!) */
+ spin_unlock(&svc->srv_lock);
+
+ spin_lock (&obd->obd_uncommitted_replies_lock);
+ /* Noop if removed already */
+ list_del_init (&rs->rs_obd_list);
+ spin_unlock (&obd->obd_uncommitted_replies_lock);
+
+ spin_lock (&exp->exp_lock);
+ /* Noop if removed already */
+ list_del_init (&rs->rs_exp_list);
+ spin_unlock (&exp->exp_lock);
+
+ spin_lock(&svc->srv_lock);
+
+ been_handled = rs->rs_handled;
+ rs->rs_handled = 1;
+
+ nlocks = rs->rs_nlocks; /* atomic "steal", but */
+ rs->rs_nlocks = 0; /* locks still on rs_locks! */
+
+ if (nlocks == 0 && !been_handled) {
+ /* If we see this, we should already have seen the warning
+ * in mds_steal_ack_locks() */
+ CWARN("All locks stolen from rs %p x"LPD64".t"LPD64
+ " o%d NID"LPX64"\n",
+ rs,
+ rs->rs_xid, rs->rs_transno,
+ rs->rs_msg.opc, exp->exp_connection->c_peer.peer_nid);
+ }
+
+ if ((!been_handled && rs->rs_on_net) ||
+ nlocks > 0) {
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ if (!been_handled && rs->rs_on_net) {
+ PtlMDUnlink(rs->rs_md_h);
+ /* Ignore return code; we're racing with
+ * completion... */
+ }
+
+ while (nlocks-- > 0)
+ ldlm_lock_decref(&rs->rs_locks[nlocks],
+ rs->rs_modes[nlocks]);
+
+ spin_lock_irqsave(&svc->srv_lock, flags);
+ }
+
+ rs->rs_scheduled = 0;
+
+ if (!rs->rs_on_net) {
+ /* Off the net */
+ svc->srv_n_difficult_replies--;
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ class_export_put (exp);
+ rs->rs_export = NULL;
+ lustre_free_reply_state (rs);
+ atomic_dec (&svc->srv_outstanding_replies);
+ RETURN(1);
+ }
+
+ /* still on the net; callback will schedule */
+ spin_unlock_irqrestore (&svc->srv_lock, flags);
+ RETURN(1);
+}
+
+#ifndef __KERNEL__
+/* FIXME make use of timeout later */
+int
+liblustre_check_services (void *arg)
+{
+ int did_something = 0;
+ struct list_head *tmp, *nxt;
+ ENTRY;
+
+ /* I'm relying on being single threaded, not to have to lock
+ * ptlrpc_all_services etc */
+ list_for_each_safe (tmp, nxt, &ptlrpc_all_services) {
+ struct ptlrpc_service *svc =
+ list_entry (tmp, struct ptlrpc_service, srv_list);
+
+ if (svc->srv_nthreads != 0) /* I've recursed */
+ continue;
+
+ /* service threads can block for bulk, so this limits us
+ * (arbitrarily) to recursing 1 stack frame per service.
+ * Note that the problem with recursion is that we have to
+ * unwind completely before our caller can resume. */
+
+ svc->srv_nthreads++;
+
+ while (ptlrpc_server_handle_reply (svc))
+ did_something++;
+
+ while (ptlrpc_server_handle_request (svc))
+ did_something++;
+
+ svc->srv_nthreads--;
+ }
- return rc;
+ RETURN(did_something);
}
+#else /* __KERNEL__ */
+
/* Don't use daemonize, it removes fs struct from new thread (bug 418) */
void ptlrpc_daemonize(void)
{
reparent_to_init();
}
-static long timeval_sub(struct timeval *large, struct timeval *small)
-{
- return (large->tv_sec - small->tv_sec) * 1000000 +
- (large->tv_usec - small->tv_usec);
-}
-
static int ptlrpc_main(void *arg)
{
- struct ptlrpc_svc_data *data = arg;
- struct obd_device *obddev = data->dev;
- struct ptlrpc_service *svc = data->svc;
- struct ptlrpc_thread *thread = data->thread;
- struct ptlrpc_request *request;
- ptl_event_t *event;
- unsigned long flags;
- struct timeval start_time, finish_time;
- long total;
- int rc = 0;
+ struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
+ struct ptlrpc_service *svc = data->svc;
+ struct ptlrpc_thread *thread = data->thread;
+ unsigned long flags;
ENTRY;
lock_kernel();
THREAD_NAME(current->comm, "%s", data->name);
unlock_kernel();
- OBD_ALLOC(event, sizeof(*event));
- if (event == NULL)
- GOTO(out, rc = -ENOMEM);
- OBD_ALLOC(request, sizeof(*request));
- if (request == NULL)
- GOTO(out_event, rc = -ENOMEM);
-
/* Record that the thread is running */
thread->t_flags = SVC_RUNNING;
wake_up(&thread->t_ctl_waitq);
+ spin_lock_irqsave(&svc->srv_lock, flags);
+ svc->srv_nthreads++;
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
/* XXX maintain a list of all managed devices: insert here */
- do_gettimeofday(&finish_time);
- /* And now, loop forever on requests */
- while (1) {
+ while ((thread->t_flags & SVC_STOPPING) == 0 ||
+ svc->srv_n_difficult_replies != 0) {
+ /* Don't exit while there are replies to be handled */
struct l_wait_info lwi = { 0 };
- l_wait_event_exclusive(svc->srv_waitq,
- ptlrpc_check_event(svc, thread, event),
- &lwi);
-
- spin_lock(&svc->srv_lock);
- if (thread->t_flags & SVC_STOPPING) {
- thread->t_flags &= ~SVC_STOPPING;
- spin_unlock(&svc->srv_lock);
-
- EXIT;
- break;
- }
-
- if (!(thread->t_flags & SVC_EVENT)) {
- CERROR("unknown flag in service");
- spin_unlock(&svc->srv_lock);
- LBUG();
- EXIT;
- break;
- }
-
- thread->t_flags &= ~SVC_EVENT;
- spin_unlock(&svc->srv_lock);
-
- do_gettimeofday(&start_time);
- total = timeval_sub(&start_time, &event->arrival_time);
- if (svc->srv_stats != NULL) {
- lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
- total);
- lprocfs_counter_add(svc->srv_stats,
- PTLRPC_SVCIDLETIME_CNTR,
- timeval_sub(&start_time,
- &finish_time));
-#if 0 /* Wait for b_eq branch */
- lprocfs_counter_add(svc->srv_stats,
- PTLRPC_SVCEQDEPTH_CNTR, 0);
-#endif
- }
- if (total / 1000000 > (long)obd_timeout) {
- CERROR("Dropping request from NID "LPX64" because it's "
- "%ld seconds old.\n", event->initiator.nid,
- total / 1000000); /* bug 1502 */
- } else {
- CDEBUG(D_HA, "request from NID "LPX64" noticed after "
- "%ldus\n", event->initiator.nid, total);
- rc = handle_incoming_request(obddev, svc, event,
- request);
- }
- do_gettimeofday(&finish_time);
- total = timeval_sub(&finish_time, &start_time);
-
- CDEBUG((total / 1000000 > (long)obd_timeout) ? D_ERROR : D_HA,
- "request "LPU64" from NID "LPX64" processed in %ldus "
- "(%ldus total)\n", request->rq_xid, event->initiator.nid,
- total, timeval_sub(&finish_time, &event->arrival_time));
-
- if (svc->srv_stats != NULL) {
- int opc = opcode_offset(request->rq_reqmsg->opc);
- if (opc > 0) {
- LASSERT(opc < LUSTRE_MAX_OPCODES);
- lprocfs_counter_add(svc->srv_stats,
- opc + PTLRPC_LAST_CNTR,
- total);
- }
- }
+ l_wait_event_exclusive (svc->srv_waitq,
+ (thread->t_flags & SVC_STOPPING) != 0 ||
+ !list_empty (&svc->srv_reply_queue) ||
+ (!list_empty (&svc->srv_request_queue) &&
+ (svc->srv_n_difficult_replies == 0 ||
+ svc->srv_n_active_reqs <
+ (svc->srv_nthreads - 1))),
+ &lwi);
+
+ if (!list_empty (&svc->srv_reply_queue))
+ ptlrpc_server_handle_reply (svc);
+
+ /* only handle requests if there are no difficult replies
+ * outstanding, or I'm not the last thread handling
+ * requests */
+ if (!list_empty (&svc->srv_request_queue) &&
+ (svc->srv_n_difficult_replies == 0 ||
+ svc->srv_n_active_reqs < (svc->srv_nthreads - 1)))
+ ptlrpc_server_handle_request (svc);
}
- /* NB should wait for all SENT callbacks to complete before exiting
- * here. Unfortunately at this time there is no way to track this
- * state. */
- OBD_FREE(request, sizeof(*request));
-out_event:
- OBD_FREE(event, sizeof(*event));
-out:
+ spin_lock_irqsave(&svc->srv_lock, flags);
+
+ svc->srv_nthreads--; /* must know immediately */
thread->t_flags = SVC_STOPPED;
wake_up(&thread->t_ctl_waitq);
- CDEBUG(D_NET, "service thread exiting, process %d: rc = %d\n",
- current->pid, rc);
- return rc;
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ CDEBUG(D_NET, "service thread exiting, process %d\n", current->pid);
+ return 0;
}
static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
struct ptlrpc_thread *thread)
{
struct l_wait_info lwi = { 0 };
+ unsigned long flags;
- spin_lock(&svc->srv_lock);
+ spin_lock_irqsave(&svc->srv_lock, flags);
thread->t_flags = SVC_STOPPING;
- spin_unlock(&svc->srv_lock);
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
wake_up_all(&svc->srv_waitq);
l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
&lwi);
+
+ spin_lock_irqsave(&svc->srv_lock, flags);
+ list_del(&thread->t_link);
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
+
+ OBD_FREE(thread, sizeof(*thread));
}
void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
{
- spin_lock(&svc->srv_lock);
+ unsigned long flags;
+ struct ptlrpc_thread *thread;
+
+ spin_lock_irqsave(&svc->srv_lock, flags);
while (!list_empty(&svc->srv_threads)) {
- struct ptlrpc_thread *thread;
- thread = list_entry(svc->srv_threads.next, struct ptlrpc_thread,
- t_link);
- spin_unlock(&svc->srv_lock);
+ thread = list_entry(svc->srv_threads.next,
+ struct ptlrpc_thread, t_link);
+
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
ptlrpc_stop_thread(svc, thread);
- spin_lock(&svc->srv_lock);
- list_del(&thread->t_link);
- OBD_FREE(thread, sizeof(*thread));
+ spin_lock_irqsave(&svc->srv_lock, flags);
}
- spin_unlock(&svc->srv_lock);
+
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
}
int ptlrpc_start_n_threads(struct obd_device *dev, struct ptlrpc_service *svc,
struct l_wait_info lwi = { 0 };
struct ptlrpc_svc_data d;
struct ptlrpc_thread *thread;
+ unsigned long flags;
int rc;
ENTRY;
if (thread == NULL)
RETURN(-ENOMEM);
init_waitqueue_head(&thread->t_ctl_waitq);
-
+
d.dev = dev;
d.svc = svc;
d.name = name;
d.thread = thread;
- spin_lock(&svc->srv_lock);
+ spin_lock_irqsave(&svc->srv_lock, flags);
list_add(&thread->t_link, &svc->srv_threads);
- spin_unlock(&svc->srv_lock);
+ spin_unlock_irqrestore(&svc->srv_lock, flags);
/* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
* just drop the VM and FILES in ptlrpc_daemonize() right away.
RETURN(0);
}
+#endif
int ptlrpc_unregister_service(struct ptlrpc_service *service)
{
- int i, rc;
+ int i;
+ int rc;
+ unsigned long flags;
struct ptlrpc_srv_ni *srv_ni;
+ struct l_wait_info lwi;
+ struct list_head *tmp;
- LASSERT (list_empty (&service->srv_threads));
+ LASSERT(list_empty(&service->srv_threads));
- /* XXX We could reply (with failure) to all buffered requests
- * _after_ unlinking _all_ the request buffers, but _before_
- * freeing them.
- */
+ spin_lock (&ptlrpc_all_services_lock);
+ list_del_init (&service->srv_list);
+ spin_unlock (&ptlrpc_all_services_lock);
+
+ for (i = 0; i < ptlrpc_ninterfaces; i++) {
+ srv_ni = &service->srv_interfaces[i];
+ CDEBUG(D_NET, "%s: tearing down interface %s\n",
+ service->srv_name, srv_ni->sni_ni->pni_name);
+
+ /* Unlink all the request buffers. This forces a 'final'
+ * event with its 'unlink' flag set for each rqbd */
+ list_for_each(tmp, &srv_ni->sni_rqbds) {
+ struct ptlrpc_request_buffer_desc *rqbd =
+ list_entry(tmp, struct ptlrpc_request_buffer_desc,
+ rqbd_list);
+
+ rc = PtlMDUnlink(rqbd->rqbd_md_h);
+ LASSERT (rc == PTL_OK || rc == PTL_INV_MD);
+ }
+
+ /* Wait for the network to release any buffers it's
+ * currently filling */
+ for (;;) {
+ spin_lock_irqsave(&service->srv_lock, flags);
+ rc = srv_ni->sni_nrqbd_receiving;
+ spin_unlock_irqrestore(&service->srv_lock, flags);
+
+ if (rc == 0)
+ break;
+
+ /* Network access will complete in finite time but
+ * the HUGE timeout lets us CWARN for visibility of
+ * sluggish NALs */
+ lwi = LWI_TIMEOUT(300 * HZ, NULL, NULL);
+ rc = l_wait_event(service->srv_waitq,
+ srv_ni->sni_nrqbd_receiving == 0,
+ &lwi);
+ if (rc == -ETIMEDOUT)
+ CWARN("Waiting for request buffers on "
+ "service %s on interface %s ",
+ service->srv_name, srv_ni->sni_ni->pni_name);
+ }
+
+ /* schedule all outstanding replies to terminate them */
+ spin_lock_irqsave(&service->srv_lock, flags);
+ while (!list_empty(&srv_ni->sni_active_replies)) {
+ struct ptlrpc_reply_state *rs =
+ list_entry(srv_ni->sni_active_replies.next,
+ struct ptlrpc_reply_state,
+ rs_list);
+ ptlrpc_schedule_difficult_reply(rs);
+ }
+ spin_unlock_irqrestore(&service->srv_lock, flags);
+ }
+
+ /* purge the request queue. NB No new replies (rqbds all unlinked)
+ * and no service threads, so I'm the only thread noodling the
+ * request queue now */
+ while (!list_empty(&service->srv_request_queue)) {
+ struct ptlrpc_request *req =
+ list_entry(service->srv_request_queue.next,
+ struct ptlrpc_request,
+ rq_list);
+
+ list_del(&req->rq_list);
+ service->srv_n_queued_reqs--;
+ req->rq_rqbd->rqbd_refcount--;
+
+ ptlrpc_free_server_req(req);
+ }
+ LASSERT(service->srv_n_queued_reqs == 0);
+ /* Now free all the request buffers since nothing references them
+ * any more... */
for (i = 0; i < ptlrpc_ninterfaces; i++) {
srv_ni = &service->srv_interfaces[i];
- CDEBUG (D_NET, "%s: tearing down interface %s\n",
- service->srv_name, srv_ni->sni_ni->pni_name);
- while (!list_empty (&srv_ni->sni_rqbds)) {
+ while (!list_empty(&srv_ni->sni_rqbds)) {
struct ptlrpc_request_buffer_desc *rqbd =
- list_entry (srv_ni->sni_rqbds.next,
- struct ptlrpc_request_buffer_desc,
- rqbd_list);
-
- list_del (&rqbd->rqbd_list);
-
- LASSERT (atomic_read (&rqbd->rqbd_refcount) > 0);
- /* refcount could be anything; it's possible for
- * the buffers to continued to get filled after all
- * the server threads exited. But we know they
- * _have_ exited.
- */
-
- (void) PtlMEUnlink(rqbd->rqbd_me_h);
- /* The callback handler could have unlinked this ME
- * already (we're racing with her) but it's safe to
- * ensure it _has_ been unlinked.
- */
-
- OBD_FREE (rqbd->rqbd_buffer, service->srv_buf_size);
- OBD_FREE (rqbd, sizeof (*rqbd));
- srv_ni->sni_nrqbds--;
+ list_entry(srv_ni->sni_rqbds.next,
+ struct ptlrpc_request_buffer_desc,
+ rqbd_list);
+
+ ptlrpc_free_rqbd(rqbd);
}
+ }
- LASSERT (srv_ni->sni_nrqbds == 0);
+ /* wait for all outstanding replies to complete (they were
+ * scheduled having been flagged to abort above) */
+ while (atomic_read(&service->srv_outstanding_replies) != 0) {
+ struct l_wait_info lwi = LWI_TIMEOUT(10 * HZ, NULL, NULL);
- if (!PtlHandleEqual (srv_ni->sni_eq_h, PTL_HANDLE_NONE)) {
- rc = PtlEQFree(srv_ni->sni_eq_h);
- if (rc)
- CERROR("%s.%d: PtlEQFree failed on %s: %d\n",
- service->srv_name, i,
- srv_ni->sni_ni->pni_name, rc);
+ rc = l_wait_event(service->srv_waitq,
+ !list_empty(&service->srv_reply_queue), &lwi);
+ LASSERT(rc == 0 || rc == -ETIMEDOUT);
+
+ if (rc == 0) {
+ ptlrpc_server_handle_reply(service);
+ continue;
}
+ CWARN("Unexpectedly long timeout %p\n", service);
}
ptlrpc_lprocfs_unregister_service(service);
OBD_FREE(service,
- offsetof (struct ptlrpc_service,
- srv_interfaces[ptlrpc_ninterfaces]));
+ offsetof(struct ptlrpc_service,
+ srv_interfaces[ptlrpc_ninterfaces]));
return 0;
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff