Whamcloud - gitweb
LU-5435 lnet: lustre network latency simulation 09/11409/14
authorLiang Zhen <liang.zhen@intel.com>
Wed, 6 Aug 2014 02:49:19 +0000 (10:49 +0800)
committerOleg Drokin <oleg.drokin@intel.com>
Sat, 1 Nov 2014 05:02:10 +0000 (05:02 +0000)
Incoming lnet message can be delayed for seconds if it can match
any of LNet Delay Rules.

User can add/remove/list Delay Rule by lctl commands:
- lctl net_delay_add
  Add a new Delay Rule to LNet, options
  <-s | --source SRC_NID>
  <-d | --dest DST_NID>
  <<-r | --rate RATE_NUMBER>
   <-i | --interlval SECONDS>>
  <-l | --latency DELAY_LATENCY>

- lctl net_delay_del
  Remove matched Delay Rule from LNet, options:
  <[-a | --all] |
   <-s | --source SRC_NID>
   <-d | --dest DST_NID>>

- lctl net_delay_list
  List all Delay Rules in LNet

- lctl net_delay_reset
  Reset statistic counters for all Delay Rules

Signed-off-by: Liang Zhen <liang.zhen@intel.com>
Change-Id: Iba1234908918ad1619ae85aa2f0ad77a992374aa
Reviewed-on: http://review.whamcloud.com/11409
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Amir Shehata <amir.shehata@intel.com>
Reviewed-by: Bobi Jam <bobijam@gmail.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
lnet/include/lnet/lib-lnet.h
lnet/include/lnet/lib-types.h
lnet/include/lnet/lnetctl.h
lnet/lnet/api-ni.c
lnet/lnet/lib-move.c
lnet/lnet/lib-msg.c
lnet/lnet/net_fault.c
lnet/utils/portals.c
lustre/utils/lctl.c

index 817963c..d0b4b51 100644 (file)
@@ -836,11 +836,22 @@ void lnet_portals_destroy(void);
 /* message functions */
 int lnet_parse (lnet_ni_t *ni, lnet_hdr_t *hdr,
                 lnet_nid_t fromnid, void *private, int rdma_req);
+int lnet_parse_local(lnet_ni_t *ni, lnet_msg_t *msg);
+int lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg);
+
 void lnet_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg, int delayed,
                unsigned int offset, unsigned int mlen, unsigned int rlen);
+void lnet_ni_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg,
+                 int delayed, unsigned int offset,
+                 unsigned int mlen, unsigned int rlen);
+
 lnet_msg_t *lnet_create_reply_msg (lnet_ni_t *ni, lnet_msg_t *get_msg);
 void lnet_set_reply_msg_len(lnet_ni_t *ni, lnet_msg_t *msg, unsigned int len);
+
 void lnet_finalize(lnet_ni_t *ni, lnet_msg_t *msg, int rc);
+
+void lnet_drop_message(lnet_ni_t *ni, int cpt, void *private,
+                      unsigned int nob);
 void lnet_drop_delayed_msg_list(struct list_head *head, char *reason);
 void lnet_recv_delayed_msg_list(struct list_head *head);
 
@@ -861,6 +872,14 @@ void lnet_fault_fini(void);
 
 bool lnet_drop_rule_match(lnet_hdr_t *hdr);
 
+int lnet_delay_rule_add(struct lnet_fault_attr *attr);
+int lnet_delay_rule_del(lnet_nid_t src, lnet_nid_t dst, bool shutdown);
+int lnet_delay_rule_list(int pos, struct lnet_fault_attr *attr,
+                        struct lnet_fault_stat *stat);
+void lnet_delay_rule_reset(void);
+void lnet_delay_rule_check(void);
+bool lnet_delay_rule_match_locked(lnet_hdr_t *hdr, struct lnet_msg *msg);
+
 /** @} lnet_fault_simulation */
 
 void lnet_counters_get(lnet_counters_t *counters);
index 8f14572..b55d598 100644 (file)
@@ -208,6 +208,7 @@ typedef struct lnet_msg {
         unsigned int          msg_rtrcredit:1;    /* taken a globel router credit */
         unsigned int          msg_peerrtrcredit:1; /* taken a peer router credit */
         unsigned int          msg_onactivelist:1; /* on the activelist */
+       unsigned int          msg_rdma_get:1;
 
         struct lnet_peer     *msg_txpeer;         /* peer I'm sending to */
         struct lnet_peer     *msg_rxpeer;         /* peer I received from */
@@ -566,6 +567,11 @@ typedef struct {
        __u32                   lrn_net;
 } lnet_remotenet_t;
 
+/** lnet message has credit and can be submitted to lnd for send/receive */
+#define LNET_CREDIT_OK         0
+/** lnet message is waiting for credit */
+#define LNET_CREDIT_WAIT       1
+
 typedef struct {
        /* my free buffer pool */
        struct list_head        rbp_bufs;
@@ -780,6 +786,7 @@ typedef struct
        /* failure simulation */
        struct list_head                ln_test_peers;
        struct list_head                ln_drop_rules;
+       struct list_head                ln_delay_rules;
 
        struct list_head                ln_nis;         /* LND instances */
        /* NIs bond on specific CPT(s) */
@@ -819,6 +826,7 @@ typedef struct
 
        struct mutex                    ln_api_mutex;
        struct mutex                    ln_lnd_mutex;
+       struct mutex                    ln_delay_mutex;
 #else
 # ifndef HAVE_LIBPTHREAD
        int                             ln_api_mutex;
index 23560e6..b42eb8a 100644 (file)
@@ -30,6 +30,10 @@ enum {
        LNET_CTL_DROP_DEL,
        LNET_CTL_DROP_RESET,
        LNET_CTL_DROP_LIST,
+       LNET_CTL_DELAY_ADD,
+       LNET_CTL_DELAY_DEL,
+       LNET_CTL_DELAY_RESET,
+       LNET_CTL_DELAY_LIST,
 };
 
 #define LNET_ACK_BIT           (1 << 0)
@@ -75,7 +79,17 @@ struct lnet_fault_attr {
                         */
                        __u32                   da_interval;
                } drop;
-               /** TODO: add more */
+               /** message latency simulation */
+               struct {
+                       __u32                   la_rate;
+                       /**
+                        * time interval of message delay, it is exclusive
+                        * with la_rate
+                        */
+                       __u32                   la_interval;
+                       /** latency to delay */
+                       __u32                   la_latency;
+               } delay;
                __u64                   space[8];
        } u;
 
@@ -98,7 +112,10 @@ struct lnet_fault_stat {
                        /** total # dropped messages */
                        __u64                   ds_dropped;
                } drop;
-               /** TODO: add more */
+               struct {
+                       /** total # delayed messages */
+                       __u64                   ls_delayed;
+               } delay;
                __u64                   space[8];
        } u;
 };
@@ -150,6 +167,10 @@ int jt_ptl_drop_add(int argc, char **argv);
 int jt_ptl_drop_del(int argc, char **argv);
 int jt_ptl_drop_reset(int argc, char **argv);
 int jt_ptl_drop_list(int argc, char **argv);
+int jt_ptl_delay_add(int argc, char **argv);
+int jt_ptl_delay_del(int argc, char **argv);
+int jt_ptl_delay_reset(int argc, char **argv);
+int jt_ptl_delay_list(int argc, char **argv);
 
 int dbg_initialize(int argc, char **argv);
 int jt_dbg_filter(int argc, char **argv);
index 5ecf560..74d5b03 100644 (file)
@@ -757,6 +757,7 @@ lnet_prepare(lnet_pid_t requested_pid)
        INIT_LIST_HEAD(&the_lnet.ln_nis_zombie);
        INIT_LIST_HEAD(&the_lnet.ln_routers);
        INIT_LIST_HEAD(&the_lnet.ln_drop_rules);
+       INIT_LIST_HEAD(&the_lnet.ln_delay_rules);
 
        rc = lnet_create_remote_nets_table();
        if (rc != 0)
index 00f6614..4211e4f 100644 (file)
 
 #include <lnet/lib-lnet.h>
 
-/** lnet message has credit and can be submitted to lnd for send/receive */
-#define LNET_CREDIT_OK         0
-/** lnet message is waiting for credit */
-#define LNET_CREDIT_WAIT       1
-
 static int local_nid_dist_zero = 1;
 CFS_MODULE_PARM(local_nid_dist_zero, "i", int, 0444,
                 "Reserved");
@@ -1472,7 +1467,7 @@ lnet_send(lnet_nid_t src_nid, lnet_msg_t *msg, lnet_nid_t rtr_nid)
        return 0; /* rc == LNET_CREDIT_OK or LNET_CREDIT_WAIT */
 }
 
-static void
+void
 lnet_drop_message(lnet_ni_t *ni, int cpt, void *private, unsigned int nob)
 {
        lnet_net_lock(cpt);
@@ -1740,7 +1735,7 @@ lnet_parse_ack(lnet_ni_t *ni, lnet_msg_t *msg)
  * \retval LNET_CREDIT_WAIT    If \a msg is blocked because w/o buffer
  * \retval -ve                 error code
  */
-static int
+int
 lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
 {
        int     rc = 0;
@@ -1768,6 +1763,33 @@ lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
        return rc;
 }
 
+int
+lnet_parse_local(lnet_ni_t *ni, lnet_msg_t *msg)
+{
+       int     rc;
+
+       switch (msg->msg_type) {
+       case LNET_MSG_ACK:
+               rc = lnet_parse_ack(ni, msg);
+               break;
+       case LNET_MSG_PUT:
+               rc = lnet_parse_put(ni, msg);
+               break;
+       case LNET_MSG_GET:
+               rc = lnet_parse_get(ni, msg, msg->msg_rdma_get);
+               break;
+       case LNET_MSG_REPLY:
+               rc = lnet_parse_reply(ni, msg);
+               break;
+       default: /* prevent an unused label if !kernel */
+               LASSERT(0);
+               return -EPROTO;
+       }
+
+       LASSERT(rc == 0 || rc == ENOENT);
+       return rc;
+}
+
 char *
 lnet_msgtyp2str (int type)
 {
@@ -2000,6 +2022,7 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
        msg->msg_type = type;
        msg->msg_private = private;
        msg->msg_receiving = 1;
+       msg->msg_rdma_get = rdma_req;
        msg->msg_len = msg->msg_wanted = payload_length;
        msg->msg_offset = 0;
        msg->msg_hdr = *hdr;
@@ -2034,6 +2057,13 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
 
        lnet_msg_commit(msg, cpt);
 
+       /* message delay simulation */
+       if (unlikely(!list_empty(&the_lnet.ln_delay_rules) &&
+                    lnet_delay_rule_match_locked(hdr, msg))) {
+               lnet_net_unlock(cpt);
+               return 0;
+       }
+
        if (!for_me) {
                rc = lnet_parse_forward_locked(ni, msg);
                lnet_net_unlock(cpt);
@@ -2050,29 +2080,10 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
 
        lnet_net_unlock(cpt);
 
-       switch (type) {
-       case LNET_MSG_ACK:
-               rc = lnet_parse_ack(ni, msg);
-               break;
-       case LNET_MSG_PUT:
-               rc = lnet_parse_put(ni, msg);
-               break;
-       case LNET_MSG_GET:
-               rc = lnet_parse_get(ni, msg, rdma_req);
-               break;
-       case LNET_MSG_REPLY:
-               rc = lnet_parse_reply(ni, msg);
-               break;
-       default:
-               LASSERT(0);
-               rc = -EPROTO;
-               goto free_drop;  /* prevent an unused label if !kernel */
-       }
-
-       if (rc == 0)
-               return 0;
-
-       LASSERT(rc == ENOENT);
+       rc = lnet_parse_local(ni, msg);
+       if (rc != 0)
+               goto free_drop;
+       return 0;
 
  free_drop:
        LASSERT(msg->msg_md == NULL);
index b1995f2..7613568 100644 (file)
@@ -537,6 +537,12 @@ lnet_finalize (lnet_ni_t *ni, lnet_msg_t *msg, int status)
                        break;
        }
 
+       if (unlikely(!list_empty(&the_lnet.ln_delay_rules))) {
+               lnet_net_unlock(cpt);
+               lnet_delay_rule_check();
+               lnet_net_lock(cpt);
+       }
+
        container->msc_finalizers[my_slot] = NULL;
        lnet_net_unlock(cpt);
 
index 1810121..661cc5d 100644 (file)
@@ -137,6 +137,10 @@ lnet_fault_stat_inc(struct lnet_fault_stat *stat, unsigned int type)
 }
 
 /**
+ * LNet message drop simulation
+ */
+
+/**
  * Add a new drop rule to LNet
  * There is no check for duplicated drop rule, all rules will be checked for
  * incoming message.
@@ -147,8 +151,10 @@ lnet_drop_rule_add(struct lnet_fault_attr *attr)
        struct lnet_drop_rule *rule;
        ENTRY;
 
-       if ((attr->u.drop.da_rate == 0) == (attr->u.drop.da_interval == 0)) {
-               CDEBUG(D_NET, "invalid rate %d or interval %d\n",
+       if (!((attr->u.drop.da_rate == 0) ^ (attr->u.drop.da_interval == 0))) {
+               CDEBUG(D_NET,
+                      "please provide either drop rate or drop interval, "
+                      "but not both at the same time %d/%d\n",
                       attr->u.drop.da_rate, attr->u.drop.da_interval);
                RETURN(-EINVAL);
        }
@@ -380,6 +386,563 @@ lnet_drop_rule_match(lnet_hdr_t *hdr)
        return drop;
 }
 
+/**
+ * LNet Delay Simulation
+ */
+/** timestamp (second) to send delayed message */
+#define msg_delay_send          msg_ev.hdr_data
+
+struct lnet_delay_rule {
+       /** link chain on the_lnet.ln_delay_rules */
+       struct list_head        dl_link;
+       /** link chain on delay_dd.dd_sched_rules */
+       struct list_head        dl_sched_link;
+       /** attributes of this rule */
+       struct lnet_fault_attr  dl_attr;
+       /** lock to protect \a below members */
+       spinlock_t              dl_lock;
+       /** refcount of delay rule */
+       atomic_t                dl_refcount;
+       /**
+        * the message sequence to delay, which means message is delayed when
+        * dl_stat.fs_count == dl_delay_at
+        */
+       unsigned long           dl_delay_at;
+       /**
+        * seconds to delay the next message, it's exclusive with dl_delay_at
+        */
+       cfs_time_t              dl_delay_time;
+       /** baseline to caculate dl_delay_time */
+       cfs_time_t              dl_time_base;
+       /** jiffies to send the next delayed message */
+       unsigned long           dl_msg_send;
+       /** delayed message list */
+       struct list_head        dl_msg_list;
+       /** statistic of delayed messages */
+       struct lnet_fault_stat  dl_stat;
+       /** timer to wakeup delay_daemon */
+       struct timer_list       dl_timer;
+};
+
+struct delay_daemon_data {
+       /** serialise rule add/remove */
+       struct mutex            dd_mutex;
+       /** protect rules on \a dd_sched_rules */
+       spinlock_t              dd_lock;
+       /** scheduled delay rules (by timer) */
+       struct list_head        dd_sched_rules;
+       /** deamon thread sleeps at here */
+       wait_queue_head_t       dd_waitq;
+       /** controler (lctl command) wait at here */
+       wait_queue_head_t       dd_ctl_waitq;
+       /** deamon is running */
+       unsigned int            dd_running;
+       /** deamon stopped */
+       unsigned int            dd_stopped;
+};
+
+static struct delay_daemon_data        delay_dd;
+
+static cfs_time_t
+round_timeout(cfs_time_t timeout)
+{
+       return cfs_time_seconds((unsigned int)
+                       cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
+}
+
+static void
+delay_rule_decref(struct lnet_delay_rule *rule)
+{
+       if (atomic_dec_and_test(&rule->dl_refcount)) {
+               LASSERT(list_empty(&rule->dl_sched_link));
+               LASSERT(list_empty(&rule->dl_msg_list));
+               LASSERT(list_empty(&rule->dl_link));
+
+               CFS_FREE_PTR(rule);
+       }
+}
+
+/**
+ * check source/destination NID, portal, message type and delay rate,
+ * decide whether should delay this message or not
+ */
+static bool
+delay_rule_match(struct lnet_delay_rule *rule, lnet_nid_t src,
+               lnet_nid_t dst, unsigned int type, unsigned int portal,
+               struct lnet_msg *msg)
+{
+       struct lnet_fault_attr  *attr = &rule->dl_attr;
+       bool                     delay;
+
+       if (!lnet_fault_attr_match(attr, src, dst, type, portal))
+               return false;
+
+       /* match this rule, check delay rate now */
+       spin_lock(&rule->dl_lock);
+       if (rule->dl_delay_time != 0) { /* time based delay */
+               cfs_time_t now = cfs_time_current();
+
+               rule->dl_stat.fs_count++;
+               delay = cfs_time_aftereq(now, rule->dl_delay_time);
+               if (delay) {
+                       if (cfs_time_after(now, rule->dl_time_base))
+                               rule->dl_time_base = now;
+
+                       rule->dl_delay_time = rule->dl_time_base +
+                                            cfs_time_seconds(cfs_rand() %
+                                               attr->u.delay.la_interval);
+                       rule->dl_time_base += cfs_time_seconds(attr->u.delay.
+                                                              la_interval);
+
+                       CDEBUG(D_NET, "Delay Rule %s->%s: next delay : "
+                                     CFS_TIME_T"\n",
+                                     libcfs_nid2str(attr->fa_src),
+                                     libcfs_nid2str(attr->fa_dst),
+                                     rule->dl_delay_time);
+               }
+
+       } else { /* rate based delay */
+               delay = rule->dl_stat.fs_count++ == rule->dl_delay_at;
+               /* generate the next random rate sequence */
+               if (rule->dl_stat.fs_count % attr->u.delay.la_rate == 0) {
+                       rule->dl_delay_at = rule->dl_stat.fs_count +
+                                           cfs_rand() % attr->u.delay.la_rate;
+                       CDEBUG(D_NET, "Delay Rule %s->%s: next delay: %lu\n",
+                              libcfs_nid2str(attr->fa_src),
+                              libcfs_nid2str(attr->fa_dst), rule->dl_delay_at);
+               }
+       }
+
+       if (!delay) {
+               spin_unlock(&rule->dl_lock);
+               return false;
+       }
+
+       /* delay this message, update counters */
+       lnet_fault_stat_inc(&rule->dl_stat, type);
+       rule->dl_stat.u.delay.ls_delayed++;
+
+       list_add_tail(&msg->msg_list, &rule->dl_msg_list);
+       msg->msg_delay_send = round_timeout(
+                       cfs_time_shift(attr->u.delay.la_latency));
+       if (rule->dl_msg_send == -1) {
+               rule->dl_msg_send = msg->msg_delay_send;
+               mod_timer(&rule->dl_timer, rule->dl_msg_send);
+       }
+
+       spin_unlock(&rule->dl_lock);
+       return true;
+}
+
+/**
+ * check if \a msg can match any Delay Rule, receiving of this message
+ * will be delayed if there is a match.
+ */
+bool
+lnet_delay_rule_match_locked(lnet_hdr_t *hdr, struct lnet_msg *msg)
+{
+       struct lnet_delay_rule  *rule;
+       lnet_nid_t               src = le64_to_cpu(hdr->src_nid);
+       lnet_nid_t               dst = le64_to_cpu(hdr->dest_nid);
+       unsigned int             typ = le32_to_cpu(hdr->type);
+       unsigned int             ptl = -1;
+
+       /* NB: called with hold of lnet_net_lock */
+
+       /* NB: if Portal is specified, then only PUT and GET will be
+        * filtered by delay rule */
+       if (typ == LNET_MSG_PUT)
+               ptl = le32_to_cpu(hdr->msg.put.ptl_index);
+       else if (typ == LNET_MSG_GET)
+               ptl = le32_to_cpu(hdr->msg.get.ptl_index);
+
+       list_for_each_entry(rule, &the_lnet.ln_delay_rules, dl_link) {
+               if (delay_rule_match(rule, src, dst, typ, ptl, msg))
+                       return true;
+       }
+
+       return false;
+}
+
+/** check out delayed messages for send */
+static void
+delayed_msg_check(struct lnet_delay_rule *rule, bool all,
+                 struct list_head *msg_list)
+{
+       struct lnet_msg *msg;
+       struct lnet_msg *tmp;
+       unsigned long    now = cfs_time_current();
+
+       if (!all && rule->dl_msg_send > now)
+               return;
+
+       spin_lock(&rule->dl_lock);
+       list_for_each_entry_safe(msg, tmp, &rule->dl_msg_list, msg_list) {
+               if (!all && msg->msg_delay_send > now)
+                       break;
+
+               msg->msg_delay_send = 0;
+               list_move_tail(&msg->msg_list, msg_list);
+       }
+
+       if (list_empty(&rule->dl_msg_list)) {
+               del_timer(&rule->dl_timer);
+               rule->dl_msg_send = -1;
+
+       } else if (!list_empty(msg_list)) {
+               /* dequeued some timedout messages, update timer for the
+                * next delayed message on rule */
+               msg = list_entry(rule->dl_msg_list.next,
+                                struct lnet_msg, msg_list);
+               rule->dl_msg_send = msg->msg_delay_send;
+               mod_timer(&rule->dl_timer, rule->dl_msg_send);
+       }
+       spin_unlock(&rule->dl_lock);
+}
+
+static void
+delayed_msg_process(struct list_head *msg_list, bool drop)
+{
+       struct lnet_msg *msg;
+
+       while (!list_empty(msg_list)) {
+               struct lnet_ni *ni;
+               int             cpt;
+               int             rc;
+
+               msg = list_entry(msg_list->next, struct lnet_msg, msg_list);
+               LASSERT(msg->msg_rxpeer != NULL);
+
+               ni = msg->msg_rxpeer->lp_ni;
+               cpt = msg->msg_rx_cpt;
+
+               list_del_init(&msg->msg_list);
+               if (drop) {
+                       rc = -ECANCELED;
+
+               } else if (!msg->msg_routing) {
+                       rc = lnet_parse_local(ni, msg);
+                       if (rc == 0)
+                               continue;
+
+               } else {
+                       lnet_net_lock(cpt);
+                       rc = lnet_parse_forward_locked(ni, msg);
+                       lnet_net_unlock(cpt);
+
+                       switch (rc) {
+                       case LNET_CREDIT_OK:
+                               lnet_ni_recv(ni, msg->msg_private, msg, 0,
+                                            0, msg->msg_len, msg->msg_len);
+                       case LNET_CREDIT_WAIT:
+                               continue;
+                       default: /* failures */
+                               break;
+                       }
+               }
+
+               lnet_drop_message(ni, cpt, msg->msg_private, msg->msg_len);
+               lnet_finalize(ni, msg, rc);
+       }
+}
+
+/**
+ * Process delayed messages for scheduled rules
+ * This function can either be called by delay_rule_daemon, or by lnet_finalise
+ */
+void
+lnet_delay_rule_check(void)
+{
+       struct lnet_delay_rule  *rule;
+       struct list_head         msgs;
+
+       INIT_LIST_HEAD(&msgs);
+       while (1) {
+               if (list_empty(&delay_dd.dd_sched_rules))
+                       break;
+
+               spin_lock_bh(&delay_dd.dd_lock);
+               if (list_empty(&delay_dd.dd_sched_rules)) {
+                       spin_unlock_bh(&delay_dd.dd_lock);
+                       break;
+               }
+
+               rule = list_entry(delay_dd.dd_sched_rules.next,
+                                 struct lnet_delay_rule, dl_sched_link);
+               list_del_init(&rule->dl_sched_link);
+               spin_unlock_bh(&delay_dd.dd_lock);
+
+               delayed_msg_check(rule, false, &msgs);
+               delay_rule_decref(rule); /* -1 for delay_dd.dd_sched_rules */
+       }
+
+       if (!list_empty(&msgs))
+               delayed_msg_process(&msgs, false);
+}
+
+/** deamon thread to handle delayed messages */
+static int
+lnet_delay_rule_daemon(void *arg)
+{
+       delay_dd.dd_running = 1;
+       wake_up(&delay_dd.dd_ctl_waitq);
+
+       while (delay_dd.dd_running) {
+               wait_event_interruptible(delay_dd.dd_waitq,
+                                        !delay_dd.dd_running ||
+                                        !list_empty(&delay_dd.dd_sched_rules));
+               lnet_delay_rule_check();
+       }
+
+       /* in case more rules have been enqueued after my last check */
+       lnet_delay_rule_check();
+       delay_dd.dd_stopped = 1;
+       wake_up(&delay_dd.dd_ctl_waitq);
+
+       return 0;
+}
+
+static void
+delay_timer_cb(unsigned long arg)
+{
+       struct lnet_delay_rule *rule = (struct lnet_delay_rule *)arg;
+
+       spin_lock_bh(&delay_dd.dd_lock);
+       if (list_empty(&rule->dl_sched_link) && delay_dd.dd_running) {
+               atomic_inc(&rule->dl_refcount);
+               list_add_tail(&rule->dl_sched_link, &delay_dd.dd_sched_rules);
+               wake_up(&delay_dd.dd_waitq);
+       }
+       spin_unlock_bh(&delay_dd.dd_lock);
+}
+
+/**
+ * Add a new delay rule to LNet
+ * There is no check for duplicated delay rule, all rules will be checked for
+ * incoming message.
+ */
+int
+lnet_delay_rule_add(struct lnet_fault_attr *attr)
+{
+       struct lnet_delay_rule *rule;
+       int                     rc = 0;
+       ENTRY;
+
+       if (!((attr->u.delay.la_rate == 0) ^
+             (attr->u.delay.la_interval == 0))) {
+               CDEBUG(D_NET,
+                      "please provide either delay rate or delay interval, "
+                      "but not both at the same time %d/%d\n",
+                      attr->u.delay.la_rate, attr->u.delay.la_interval);
+               RETURN(-EINVAL);
+       }
+
+       if (attr->u.delay.la_latency == 0) {
+               CDEBUG(D_NET, "delay latency cannot be zero\n");
+               RETURN(-EINVAL);
+       }
+
+       if (lnet_fault_attr_validate(attr) != 0)
+               RETURN(-EINVAL);
+
+       CFS_ALLOC_PTR(rule);
+       if (rule == NULL)
+               RETURN(-ENOMEM);
+
+       mutex_lock(&delay_dd.dd_mutex);
+       if (!delay_dd.dd_running) {
+               struct task_struct *task;
+
+               /* NB: although LND threads will process delayed message
+                * in lnet_finalize, but there is no guarantee that LND
+                * threads will be waken up if no other message needs to
+                * be handled.
+                * Only one daemon thread, performance is not the concern
+                * of this simualation module.
+                */
+               task = kthread_run(lnet_delay_rule_daemon, NULL, "lnet_dd");
+               if (IS_ERR(task)) {
+                       rc = PTR_ERR(task);
+                       GOTO(failed, rc);
+               }
+               wait_event(delay_dd.dd_ctl_waitq, delay_dd.dd_running);
+       }
+
+       init_timer(&rule->dl_timer);
+       rule->dl_timer.function = delay_timer_cb;
+       rule->dl_timer.data = (unsigned long)rule;
+
+       spin_lock_init(&rule->dl_lock);
+       INIT_LIST_HEAD(&rule->dl_msg_list);
+       INIT_LIST_HEAD(&rule->dl_sched_link);
+
+       rule->dl_attr = *attr;
+       if (attr->u.delay.la_interval != 0) {
+               rule->dl_time_base = cfs_time_shift(attr->u.delay.la_interval);
+               rule->dl_delay_time = cfs_time_shift(cfs_rand() %
+                                                    attr->u.delay.la_interval);
+       } else {
+               rule->dl_delay_at = cfs_rand() % attr->u.delay.la_rate;
+       }
+
+       rule->dl_msg_send = -1;
+
+       lnet_net_lock(LNET_LOCK_EX);
+       atomic_set(&rule->dl_refcount, 1);
+       list_add(&rule->dl_link, &the_lnet.ln_delay_rules);
+       lnet_net_unlock(LNET_LOCK_EX);
+
+       CDEBUG(D_NET, "Added delay rule: src %s, dst %s, rate %d\n",
+              libcfs_nid2str(attr->fa_src), libcfs_nid2str(attr->fa_src),
+              attr->u.delay.la_rate);
+
+       mutex_unlock(&delay_dd.dd_mutex);
+       RETURN(0);
+ failed:
+       mutex_unlock(&delay_dd.dd_mutex);
+       CFS_FREE_PTR(rule);
+       return rc;
+}
+
+/**
+ * Remove matched Delay Rules from lnet, if \a shutdown is true or both \a src
+ * and \a dst are zero, all rules will be removed, otherwise only matched rules
+ * will be removed.
+ * If \a src is zero, then all rules have \a dst as destination will be remove
+ * If \a dst is zero, then all rules have \a src as source will be removed
+ *
+ * When a delay rule is removed, all delayed messages of this rule will be
+ * processed immediately.
+ */
+int
+lnet_delay_rule_del(lnet_nid_t src, lnet_nid_t dst, bool shutdown)
+{
+       struct lnet_delay_rule *rule;
+       struct lnet_delay_rule  *tmp;
+       struct list_head        rule_list;
+       struct list_head        msg_list;
+       int                     n = 0;
+       bool                    cleanup;
+       ENTRY;
+
+       INIT_LIST_HEAD(&rule_list);
+       INIT_LIST_HEAD(&msg_list);
+
+       if (shutdown)
+               src = dst = 0;
+
+       mutex_lock(&delay_dd.dd_mutex);
+       lnet_net_lock(LNET_LOCK_EX);
+
+       list_for_each_entry_safe(rule, tmp, &the_lnet.ln_delay_rules, dl_link) {
+               if (rule->dl_attr.fa_src != src && src != 0)
+                       continue;
+
+               if (rule->dl_attr.fa_dst != dst && dst != 0)
+                       continue;
+
+               CDEBUG(D_NET, "Remove delay rule: src %s->dst: %s (1/%d, %d)\n",
+                      libcfs_nid2str(rule->dl_attr.fa_src),
+                      libcfs_nid2str(rule->dl_attr.fa_dst),
+                      rule->dl_attr.u.delay.la_rate,
+                      rule->dl_attr.u.delay.la_interval);
+               /* refcount is taken over by rule_list */
+               list_move(&rule->dl_link, &rule_list);
+       }
+
+       /* check if we need to shutdown delay_daemon */
+       cleanup = list_empty(&the_lnet.ln_delay_rules) &&
+                 !list_empty(&rule_list);
+       lnet_net_unlock(LNET_LOCK_EX);
+
+       list_for_each_entry_safe(rule, tmp, &rule_list, dl_link) {
+               list_del_init(&rule->dl_link);
+
+               del_timer_sync(&rule->dl_timer);
+               delayed_msg_check(rule, true, &msg_list);
+               delay_rule_decref(rule); /* -1 for the_lnet.ln_delay_rules */
+               n++;
+       }
+
+       if (cleanup) { /* no more delay rule, shutdown delay_daemon */
+               LASSERT(delay_dd.dd_running);
+               delay_dd.dd_running = 0;
+               wake_up(&delay_dd.dd_waitq);
+
+               while (!delay_dd.dd_stopped)
+                       wait_event(delay_dd.dd_ctl_waitq, delay_dd.dd_stopped);
+       }
+       mutex_unlock(&delay_dd.dd_mutex);
+
+       if (!list_empty(&msg_list))
+               delayed_msg_process(&msg_list, shutdown);
+
+       RETURN(n);
+}
+
+/**
+ * List Delay Rule at position of \a pos
+ */
+int
+lnet_delay_rule_list(int pos, struct lnet_fault_attr *attr,
+                   struct lnet_fault_stat *stat)
+{
+       struct lnet_delay_rule *rule;
+       int                     cpt;
+       int                     i = 0;
+       int                     rc = -ENOENT;
+       ENTRY;
+
+       cpt = lnet_net_lock_current();
+       list_for_each_entry(rule, &the_lnet.ln_delay_rules, dl_link) {
+               if (i++ < pos)
+                       continue;
+
+               spin_lock(&rule->dl_lock);
+               *attr = rule->dl_attr;
+               *stat = rule->dl_stat;
+               spin_unlock(&rule->dl_lock);
+               rc = 0;
+               break;
+       }
+
+       lnet_net_unlock(cpt);
+       RETURN(rc);
+}
+
+/**
+ * reset counters for all Delay Rules
+ */
+void
+lnet_delay_rule_reset(void)
+{
+       struct lnet_delay_rule *rule;
+       int                     cpt;
+       ENTRY;
+
+       cpt = lnet_net_lock_current();
+
+       list_for_each_entry(rule, &the_lnet.ln_delay_rules, dl_link) {
+               struct lnet_fault_attr *attr = &rule->dl_attr;
+
+               spin_lock(&rule->dl_lock);
+
+               memset(&rule->dl_stat, 0, sizeof(rule->dl_stat));
+               if (attr->u.delay.la_rate != 0) {
+                       rule->dl_delay_at = cfs_rand() % attr->u.delay.la_rate;
+               } else {
+                       rule->dl_delay_time = cfs_time_shift(cfs_rand() %
+                                               attr->u.delay.la_interval);
+                       rule->dl_time_base = cfs_time_shift(attr->u.delay.
+                                                                 la_interval);
+               }
+               spin_unlock(&rule->dl_lock);
+       }
+
+       lnet_net_unlock(cpt);
+       EXIT;
+}
+
 int
 lnet_fault_ctl(int opc, struct libcfs_ioctl_data *data)
 {
@@ -416,6 +979,31 @@ lnet_fault_ctl(int opc, struct libcfs_ioctl_data *data)
                        return -EINVAL;
 
                return lnet_drop_rule_list(data->ioc_count, attr, stat);
+
+       case LNET_CTL_DELAY_ADD:
+               if (attr == NULL)
+                       return -EINVAL;
+
+               return lnet_delay_rule_add(attr);
+
+       case LNET_CTL_DELAY_DEL:
+               if (attr == NULL)
+                       return -EINVAL;
+
+               data->ioc_count = lnet_delay_rule_del(attr->fa_src,
+                                                     attr->fa_dst, false);
+               return 0;
+
+       case LNET_CTL_DELAY_RESET:
+               lnet_delay_rule_reset();
+               return 0;
+
+       case LNET_CTL_DELAY_LIST:
+               stat = (struct lnet_fault_stat *)data->ioc_inlbuf2;
+               if (attr == NULL || stat == NULL)
+                       return -EINVAL;
+
+               return lnet_delay_rule_list(data->ioc_count, attr, stat);
        }
 }
 
@@ -427,6 +1015,12 @@ lnet_fault_init(void)
        CLASSERT(LNET_GET_BIT == 1 << LNET_MSG_GET);
        CLASSERT(LNET_REPLY_BIT == 1 << LNET_MSG_REPLY);
 
+       mutex_init(&delay_dd.dd_mutex);
+       spin_lock_init(&delay_dd.dd_lock);
+       init_waitqueue_head(&delay_dd.dd_waitq);
+       init_waitqueue_head(&delay_dd.dd_ctl_waitq);
+       INIT_LIST_HEAD(&delay_dd.dd_sched_rules);
+
        return 0;
 }
 
@@ -434,6 +1028,9 @@ void
 lnet_fault_fini(void)
 {
        lnet_drop_rule_del(0, 0);
+       lnet_delay_rule_del(0, 0, true);
 
        LASSERT(list_empty(&the_lnet.ln_drop_rules));
+       LASSERT(list_empty(&the_lnet.ln_delay_rules));
+       LASSERT(list_empty(&delay_dd.dd_sched_rules));
 }
index 8f1b2c5..a2a9f31 100644 (file)
@@ -1355,6 +1355,7 @@ fault_simul_rule_add(__u32 opc, char *name, int argc, char **argv)
                {"dest",        required_argument,      0,      'd'},
                {"rate",        required_argument,      0,      'r'},
                {"interval",    required_argument,      0,      'i'},
+               {"latency",     required_argument,      0,      'l'},
                {"portal",      required_argument,      0,      'p'},
                {"message",     required_argument,      0,      'm'},
                {0, 0, 0, 0}
@@ -1366,7 +1367,7 @@ fault_simul_rule_add(__u32 opc, char *name, int argc, char **argv)
                return -1;
        }
 
-       optstr = "s:d:r:i:p:m:";
+       optstr = opc == LNET_CTL_DROP_ADD ? "s:d:r:i:p:m:" : "s:d:r:l:p:m:";
        memset(&attr, 0, sizeof(attr));
        while (1) {
                char c = getopt_long(argc, argv, optstr, opts, NULL);
@@ -1388,11 +1389,23 @@ fault_simul_rule_add(__u32 opc, char *name, int argc, char **argv)
                        break;
 
                case 'r': /* drop rate */
-                       attr.u.drop.da_rate = strtoul(optarg, NULL, 0);
+                       if (opc == LNET_CTL_DROP_ADD)
+                               attr.u.drop.da_rate = strtoul(optarg, NULL, 0);
+                       else
+                               attr.u.delay.la_rate = strtoul(optarg, NULL, 0);
                        break;
 
                case 'i': /* time interval (# seconds) for message drop */
-                       attr.u.drop.da_interval = strtoul(optarg, NULL, 0);
+                       if (opc == LNET_CTL_DROP_ADD)
+                               attr.u.drop.da_interval = strtoul(optarg,
+                                                                 NULL, 0);
+                       else
+                               attr.u.delay.la_interval = strtoul(optarg,
+                                                                  NULL, 0);
+                       break;
+
+               case 'l': /* seconds to wait before activating rule */
+                       attr.u.delay.la_latency = strtoul(optarg, NULL, 0);
                        break;
 
                case 'p': /* portal to filter */
@@ -1415,9 +1428,28 @@ fault_simul_rule_add(__u32 opc, char *name, int argc, char **argv)
        }
        optind = 1;
 
-       if ((attr.u.drop.da_rate == 0) == (attr.u.drop.da_interval == 0)) {
-               fprintf(stderr, "please provide drop rate or interval\n");
-               return -1;
+       if (opc == LNET_CTL_DROP_ADD) {
+               /* NB: drop rate and interval are exclusive to each other */
+               if (!((attr.u.drop.da_rate == 0) ^
+                     (attr.u.drop.da_interval == 0))) {
+                       fprintf(stderr,
+                               "please provide either drop rate or interval "
+                               "but not both at the same time.\n");
+                       return -1;
+               }
+       } else if (opc == LNET_CTL_DELAY_ADD) {
+               if (!((attr.u.delay.la_rate == 0) ^
+                     (attr.u.delay.la_interval == 0))) {
+                       fprintf(stderr,
+                               "please provide either delay rate or interval "
+                               "but not both at the same time.\n");
+                       return -1;
+               }
+
+               if (attr.u.delay.la_latency == 0) {
+                       fprintf(stderr, "latency cannot be zero\n");
+                       return -1;
+               }
        }
 
        if (attr.fa_src == 0 || attr.fa_dst == 0) {
@@ -1443,8 +1475,9 @@ fault_simul_rule_add(__u32 opc, char *name, int argc, char **argv)
        }
 
        printf("Added %s rule %s->%s (1/%d)\n",
-              name, libcfs_nid2str(attr.fa_src),
-              libcfs_nid2str(attr.fa_dst), attr.u.drop.da_rate);
+              name, libcfs_nid2str(attr.fa_src), libcfs_nid2str(attr.fa_dst),
+              opc == LNET_CTL_DROP_ADD ?
+              attr.u.drop.da_rate : attr.u.delay.la_rate);
        return 0;
 
 getopt_failed:
@@ -1458,6 +1491,12 @@ jt_ptl_drop_add(int argc, char **argv)
        return fault_simul_rule_add(LNET_CTL_DROP_ADD, "drop", argc, argv);
 }
 
+int
+jt_ptl_delay_add(int argc, char **argv)
+{
+       return fault_simul_rule_add(LNET_CTL_DELAY_ADD, "delay", argc, argv);
+}
+
 static int
 fault_simul_rule_del(__u32 opc, char *name, int argc, char **argv)
 {
@@ -1541,6 +1580,12 @@ jt_ptl_drop_del(int argc, char **argv)
        return fault_simul_rule_del(LNET_CTL_DROP_DEL, "drop", argc, argv);
 }
 
+int
+jt_ptl_delay_del(int argc, char **argv)
+{
+       return fault_simul_rule_del(LNET_CTL_DELAY_DEL, "delay", argc, argv);
+}
+
 static int
 fault_simul_rule_reset(__u32 opc, char *name, int argc, char **argv)
 {
@@ -1565,6 +1610,13 @@ jt_ptl_drop_reset(int argc, char **argv)
        return fault_simul_rule_reset(LNET_CTL_DROP_RESET, "drop", argc, argv);
 }
 
+int
+jt_ptl_delay_reset(int argc, char **argv)
+{
+       return fault_simul_rule_reset(LNET_CTL_DELAY_RESET, "delay",
+                                     argc, argv);
+}
+
 static int
 fault_simul_rule_list(__u32 opc, char *name, int argc, char **argv)
 {
@@ -1608,6 +1660,19 @@ fault_simul_rule_list(__u32 opc, char *name, int argc, char **argv)
                               stat.u.drop.ds_dropped, stat.fs_count,
                               stat.fs_put, stat.fs_ack,
                               stat.fs_get, stat.fs_reply);
+
+               } else if (opc == LNET_CTL_DELAY_LIST) {
+                       printf("%s->%s (1/%d | %d, latency %d) ptl "LPX64
+                              ", msg %x, "LPU64"/"LPU64", PUT "LPU64
+                              ", ACK "LPU64", GET "LPU64", REP "LPU64"\n",
+                              libcfs_nid2str(attr.fa_src),
+                              libcfs_nid2str(attr.fa_dst),
+                              attr.u.delay.la_rate, attr.u.delay.la_interval,
+                              attr.u.delay.la_latency,
+                              attr.fa_ptl_mask, attr.fa_msg_mask,
+                              stat.u.delay.ls_delayed, stat.fs_count,
+                              stat.fs_put, stat.fs_ack, stat.fs_get,
+                              stat.fs_reply);
                }
        }
        printf("found total %d\n", pos);
@@ -1621,6 +1686,12 @@ jt_ptl_drop_list(int argc, char **argv)
        return fault_simul_rule_list(LNET_CTL_DROP_LIST, "drop", argc, argv);
 }
 
+int
+jt_ptl_delay_list(int argc, char **argv)
+{
+       return fault_simul_rule_list(LNET_CTL_DELAY_LIST, "delay", argc, argv);
+}
+
 double
 get_cycles_per_usec ()
 {
index 6f50454..65c2b85 100644 (file)
@@ -127,6 +127,22 @@ command_t cmdlist[] = {
         "usage: net_drop_reset"},
        {"net_drop_list", jt_ptl_drop_list, 0, "list LNet drop rules\n"
         "usage: net_drop_list"},
+       {"net_delay_add", jt_ptl_delay_add, 0, "Add LNet delay rule\n"
+        "usage: net_delay_add <-s | --source NID>\n"
+        "                     <-d | --dest NID>\n"
+        "                     <<-r | --rate DROP_RATE> |\n"
+        "                      <-i | --interval SECONDS>>\n"
+        "                     <-l | --latency SECONDS>\n"
+        "                     [<-p | --portal> PORTAL...]\n"
+        "                     [<-m | --message> <PUT|ACK|GET|REPLY>...]\n"},
+       {"net_delay_del", jt_ptl_delay_del, 0, "remove LNet delay rule\n"
+        "usage: net_delay_del <[-a | --all] |\n"
+        "                     <-s | --source NID>\n"
+        "                     <-d | --dest NID>>\n"},
+       {"net_delay_reset", jt_ptl_delay_reset, 0, "reset delay rule stats\n"
+        "usage: net_delay_reset"},
+       {"net_delay_list", jt_ptl_delay_list, 0, "list LNet delay rules\n"
+        "usage: net_delay_list"},
 
         /* Device selection commands */
         {"==== obd device selection ====", jt_noop, 0, "device selection"},