diff --git a/lustre/include/lu_object.h b/lustre/include/lu_object.h
index 2b9424f..216f813 100644
--- a/lustre/include/lu_object.h
+++ b/lustre/include/lu_object.h
@@ -991,6 +991,7 @@ enum lu_context_tag {
          */
         LCT_SESSION   = 1 << 4,
 
+
         /**
          * Set when at least one of keys, having values in this context has
          * non-NULL lu_context_key::lct_exit() method. This is used to
diff --git a/lustre/include/lustre_net.h b/lustre/include/lustre_net.h
index eaa7c8c..2e0def6 100644
--- a/lustre/include/lustre_net.h
+++ b/lustre/include/lustre_net.h
@@ -470,14 +470,23 @@ enum {
         PTLRPC_NRS_TYPE_FIFO            = 1,
         PTLRPC_NRS_TYPE_CRR             = 2,
         PTLRPC_NRS_TYPE_CRR2            = 3,
+        PTLRPC_NRS_TYPE_ORR             = 4,
 };
 
 enum {
         PTLRPC_NRS_CTL_GET_INFO         = 1,
         PTLRPC_NRS_CTL_START            = 2,
         PTLRPC_NRS_CTL_STOP             = 3,
+        PTLRPC_NRS_CTL_ORR_RD_QUANTUM   = 4,
+        PTLRPC_NRS_CTL_ORR_WR_QUANTUM   = 5,
+        PTLRPC_NRS_CTL_ORR_RD_OFF_TYPE  = 6,
+        PTLRPC_NRS_CTL_ORR_WR_OFF_TYPE  = 7,
+        PTLRPC_NRS_CTL_ORR_RD_TRR_MODE  = 8,
+        PTLRPC_NRS_CTL_ORR_WR_TRR_MODE  = 9,
+        PTLRPC_NRS_CTL_ORR_RD_SUPP_REQ  = 10,
+        PTLRPC_NRS_CTL_ORR_WR_SUPP_REQ  = 11,
         /** reserved, recycle resource for inactive policy */
-        PTLRPC_NRS_CTL_SHRINK           = 9,
+        PTLRPC_NRS_CTL_SHRINK           = 12,
 };
 
 enum {
@@ -540,6 +549,7 @@ typedef struct ptlrpc_nrs_policy_desc {
 extern ptlrpc_nrs_policy_desc_t ptlrpc_nrs_fifo_desc;
 extern ptlrpc_nrs_policy_desc_t ptlrpc_nrs_crr_desc;
 extern ptlrpc_nrs_policy_desc_t ptlrpc_nrs_crr2_desc;
+extern ptlrpc_nrs_policy_desc_t ptlrpc_nrs_orr_desc;
 
 enum {
         /** already stopped */
@@ -666,41 +676,88 @@ typedef struct nrs_crr2_client {
 /**
  * Object Round Robin (reserved)
  */
+typedef __u16 nrs_orr_quantum_t;
+
+typedef struct nrs_orr_req_range {
+        __u64           or_start;
+        __u64           or_end;
+} nrs_orr_req_range_t;
+
+/* RPC types supported by NRS ORR */
+typedef enum {
+        NOS_OST_READ  = (1 << 0),
+        NOS_OST_WRITE = (1 << 1),
+        NOS_OST_RW = (1 << 2)
+} nrs_orr_supp_t;
+
+#define NOS_DFLT        NOS_OST_READ
+
+/* For unique keys for the objects, we can use OST index + object ID; later,
+ * this can be replaced by OST FID */
+typedef struct nrs_orr_key {
+        /* OST index */
+        int             ok_idx;
+        /* Object ID */
+        obd_id          ok_id;
+} nrs_orr_key_t;
+
 /**
  * private data structure for ORR binheap NRS
  */
-typedef struct nrs_orr_target {
-        ptlrpc_nrs_resource_t   ot_res;
-        cfs_binheap_t          *ot_binheap;
-        __u64                   ot_round;
-        __u64                   ot_sequence;
-        unsigned                ot_quantum;
-} nrs_orr_target_t;
+typedef struct nrs_orr_data {
+        __u64                   od_round;
+        cfs_binheap_t          *od_binheap;
+        cfs_hash_t             *od_obj_hash;
+        cfs_mem_cache_t        *od_cache;
+        ptlrpc_nrs_resource_t   od_res;
+        nrs_orr_supp_t          od_supp;
+        nrs_orr_quantum_t       od_quantum;
+        __u8                    od_physical;
+        /* When set, object IDs will be ignored */
+        __u8                    od_trr_mode;
+} nrs_orr_data_t;
 
 /**
  * target object for ORR binheap NRS which is embedded in object structure
+ * TODO: Maybe we can delete these when the corresponding backend-fs object is
+ * deleted; look at this after testing
  */
 typedef struct nrs_orr_obj {
-        ptlrpc_nrs_resource_t   co_res;
-        __u64                   co_round;
-        /** reserved for object round-robin */
-        __u64                   co_sequence;
-        /** reserved for object round-robin */
-        unsigned                co_quantum;
+        ptlrpc_nrs_resource_t   oo_res;
+        __u64                   oo_round;
+        nrs_orr_key_t           oo_key;
+        cfs_atomic_t            oo_ref;
+        cfs_hlist_node_t        oo_hnode;
+        nrs_orr_quantum_t       oo_quantum;
+        __u8                    oo_started;
 } nrs_orr_object_t;
 
 /**
  * embed it to ptlrpc request for ORR binheap NRS
  */
 typedef struct nrs_orr_req {
-        /** request header, must be the first member of structure */
         cfs_binheap_node_t      or_node;
-        /** more keys */
-        __u64                   or_start;
-        /** more keys */
-        __u64                   or_end;
+        nrs_orr_req_range_t     or_range;
+        __u64                   or_round;
+        /* Set for OST_WRITE requests */
+        __u8                    or_write;
+        /** For debugging purposes */
+        nrs_orr_key_t           or_key;
 } nrs_orr_req_t;
 
+typedef struct nrs_orr_info {
+        nrs_orr_quantum_t       oi_quantum_hp;
+        nrs_orr_quantum_t       oi_quantum_reg;
+        /* Is the offset type physical or logical */
+        __u8                    oi_physical_hp;
+        __u8                    oi_physical_reg;
+        /* Is TRR mode on or off?                 */
+        __u8                    oi_trr_mode_reg;
+        __u8                    oi_trr_mode_hp;
+        nrs_orr_supp_t          oi_supp_req_reg;
+        nrs_orr_supp_t          oi_supp_req_hp;
+} nrs_orr_info_t;
+
 /**
  *
  */
diff --git a/lustre/ptlrpc/lproc_ptlrpc.c b/lustre/ptlrpc/lproc_ptlrpc.c
index 34bd978..874d142 100644
--- a/lustre/ptlrpc/lproc_ptlrpc.c
+++ b/lustre/ptlrpc/lproc_ptlrpc.c
@@ -376,6 +376,7 @@ ptlrpc_lprocfs_rd_nrs(char *page, char **start, off_t off,
         struct ptlrpc_service *svc = data;
         ptlrpc_nrs_policy_info_t info;
         int     rc;
+        int     rc2;
 
         /* XXX it's temporary workaround, FIXME */
         rc = snprintf(page, count, "name\ttype\tstate\tfb\tqueued\tactive\n");
@@ -402,6 +403,15 @@ ptlrpc_lprocfs_rd_nrs(char *page, char **start, off_t off,
                        info.pi_name, info.pi_type, info.pi_state,
                        info.pi_fallback, (int)info.pi_req_queued,
                        (int)info.pi_req_started);
+        rc2 = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
+                                        PTLRPC_NRS_TYPE_ORR,
+                                        PTLRPC_NRS_CTL_GET_INFO, &info);
+        /* Not all policies will have ORR registered */
+        if (rc2 == 0)
+                rc += snprintf(page + rc, count, "%s\t%-8d%-8d%-8d%-8d%-8d\n",
+                               info.pi_name, info.pi_type, info.pi_state,
+                               info.pi_fallback, (int)info.pi_req_queued,
+                               (int)info.pi_req_started);
         return rc;
 }
 
@@ -437,12 +447,270 @@ ptlrpc_lprocfs_wr_nrs(struct file *file, const char *buffer,
                 type = PTLRPC_NRS_TYPE_CRR2;
                 opc = PTLRPC_NRS_CTL_START;
                 break;
+        case 30:
+                type = PTLRPC_NRS_TYPE_ORR;
+                opc = PTLRPC_NRS_CTL_STOP;
+                break;
+        case 31:
+                type = PTLRPC_NRS_TYPE_ORR;
+                opc = PTLRPC_NRS_CTL_START;
+                break;
         }
 
         ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_BOTH, type, opc, NULL);
         return count;
 }
 
+static int
+ptlrpc_lprocfs_rd_nrs_orr_quantum(char *page, char **start, off_t off,
+                                  int count, int *eof, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        nrs_orr_info_t          info;
+        int                     rc;
+
+        rc = snprintf(page, count, "num RPCs:");
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_RD_QUANTUM, &info);
+        rc += snprintf(page + rc, count, "\t%-8d\n",
+                       policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG ?
+                       info.oi_quantum_reg : info.oi_quantum_hp);
+
+        return rc;
+}
+
+/* XXX: This is only temporarily here; libcfs should really have these,
+ *      right? Keep this in line with nrs_orr_quantum_t */
+#ifndef USHORT_MAX
+# define USHORT_MAX     ((__u16)(~0U))
+#endif
+
+/* Changing the quantum only takes effect from the next round of request merges
+ * for the object */
+static int
+ptlrpc_lprocfs_wr_nrs_orr_quantum(struct file *file, const char *buffer,
+                                  unsigned long count, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        int                     val;
+
+        lprocfs_write_helper(buffer, count, &val);
+
+        if (val < 0)
+               return -EINVAL;
+        if (val > USHORT_MAX)
+                return -EINVAL;
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_WR_QUANTUM, &val);
+
+        return count;
+}
+
+/* Read the NRS ORR policy offset type, {physical|logical}
+ * TODO: These are all temporary, there is not much point in going through
+ * ptlrpc_nrs_policy_control if we already have the policy pointer here
+ */
+static int
+ptlrpc_lprocfs_rd_nrs_orr_off_type(char *page, char **start, off_t off,
+                                   int count, int *eof, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        nrs_orr_info_t          info;
+        int                     rc;
+        __u8                    physical = 0;
+
+
+        LASSERT(policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG ||
+                policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_HP);
+
+        rc = snprintf(page, count, "offset type:");
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_RD_OFF_TYPE, &info);
+
+        if (policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG) {
+                if (info.oi_physical_reg == 1)
+                        physical = 1;
+        } else {
+                if (info.oi_physical_hp == 1)
+                        physical = 1;
+        }
+
+        rc += snprintf(page + rc, count, "\t%s\n",
+                       physical == 1 ? "physical\n": "logical\n");
+
+        return rc;
+}
+
+/* Change the NRS ORR offset type between {physical|logical} */
+static int
+ptlrpc_lprocfs_wr_nrs_orr_off_type(struct file *file, const char *buffer,
+                                   unsigned long count, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        char                    off_type[9];
+
+        if (count > sizeof(off_type) - 1)
+                return -EINVAL;
+        if (cfs_copy_from_user(off_type, buffer, count))
+                return -EFAULT;
+        off_type[count] = '\0';
+
+        if (strcmp(off_type, "physical") && strcmp(off_type, "logical"))
+                return -EINVAL;
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_WR_OFF_TYPE, off_type);
+
+        return count;
+}
+
+/* Read whether the policy is operating Target-Based RR mode, by ignoring
+ * object IDs
+ */
+static int
+ptlrpc_lprocfs_rd_nrs_orr_trr_mode(char *page, char **start, off_t off,
+                                   int count, int *eof, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        nrs_orr_info_t          info;
+        int                     rc;
+        __u8                    trr_mode = 0;
+
+
+        LASSERT(policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG ||
+                policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_HP);
+
+        rc = snprintf(page, count, "TRR mode:");
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_RD_TRR_MODE, &info);
+
+        if (policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG) {
+                if (info.oi_trr_mode_reg == 1)
+                        trr_mode = 1;
+        } else {
+                if (info.oi_trr_mode_hp == 1)
+                        trr_mode = 1;
+        }
+
+        rc += snprintf(page + rc, count, "\t%s\n",
+                       trr_mode == 1 ? "on\n": "off\n");
+
+        return rc;
+}
+
+/* Change the NRS ORR offset type between {physical|logical} */
+static int
+ptlrpc_lprocfs_wr_nrs_orr_trr_mode(struct file *file, const char *buffer,
+                                   unsigned long count, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        char                    trr_mode[4];
+
+        if (count > sizeof(trr_mode) - 1)
+                return -EINVAL;
+        if (cfs_copy_from_user(trr_mode, buffer, count))
+                return -EFAULT;
+        trr_mode[count] = '\0';
+
+        if (strcmp(trr_mode, "on") && strcmp(trr_mode, "off"))
+                return -EINVAL;
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_WR_TRR_MODE, trr_mode);
+
+        return count;
+}
+
+#define NOS_OST_READ_STRING "OST_READ"
+#define NOS_OST_WRITE_STRING "OST_WRITE"
+#define NOS_OST_RW_STRING "OST_READ and OST_WRITE"
+
+/* Obtain the currectly handled type of RPCs for an ORR/TRR policy
+ */
+static int
+ptlrpc_lprocfs_rd_nrs_orr_supp_req(char *page, char **start, off_t off,
+                                   int count, int *eof, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        nrs_orr_info_t          info;
+        int                     rc;
+        nrs_orr_supp_t          supp;
+        char                    supp_string[23];
+
+
+        LASSERT(policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG ||
+                policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_HP);
+
+        rc = snprintf(page, count, "RPCs handled:");
+
+        /* TODO: Check for return values of this */
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_RD_SUPP_REQ, &info);
+
+        if (policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG) {
+                supp = info.oi_supp_req_reg;
+        } else {
+                supp = info.oi_supp_req_hp;
+        }
+
+        if (supp == NOS_OST_READ)
+                memcpy(supp_string, NOS_OST_READ_STRING, sizeof(NOS_OST_READ_STRING));
+        else if (supp == NOS_OST_WRITE)
+                memcpy(supp_string, NOS_OST_WRITE_STRING, sizeof(NOS_OST_WRITE_STRING));
+        else
+                memcpy(supp_string, NOS_OST_RW_STRING, sizeof(NOS_OST_RW_STRING));
+
+        rc += snprintf(page + rc, count, "\t%s\n",
+                       supp_string);
+
+        return rc;
+}
+
+/* Set the currectly handled type of RPCs for an ORR/TRR policy; see
+ * nrs_orr_supp_t; inputs are read, write, readwrite
+ */
+static int
+ptlrpc_lprocfs_wr_nrs_orr_supp_req(struct file *file, const char *buffer,
+                                   unsigned long count, void *data)
+{
+        ptlrpc_nrs_policy_t    *policy = data;
+        struct ptlrpc_service  *svc = policy->pol_nrs->nrs_service;
+        char                    supp[6];
+
+        if (count > sizeof(supp) - 1)
+                return -EINVAL;
+        if (cfs_copy_from_user(supp, buffer, count))
+                return -EFAULT;
+        supp[count] = '\0';
+
+        if (strcmp(supp, "read") && strcmp(supp, "write") &&
+            strcmp(supp, "readwrite"))
+                return -EINVAL;
+
+        ptlrpc_nrs_policy_control(svc, policy->pol_nrs->nrs_queue_type,
+                                  PTLRPC_NRS_TYPE_ORR,
+                                  PTLRPC_NRS_CTL_ORR_WR_SUPP_REQ, supp);
+
+        return count;
+}
+
 struct ptlrpc_srh_iterator {
         __u64                  srhi_seq;
         struct ptlrpc_request *srhi_req;
@@ -720,7 +988,7 @@ void ptlrpc_lprocfs_register_service(struct proc_dir_entry *entry,
                 {.name       = "timeouts",
                  .read_fptr  = ptlrpc_lprocfs_rd_timeouts,
                  .data       = svc},
-                {.name       = "nrs_polices",
+                {.name       = "nrs_policies",
                  .read_fptr  = ptlrpc_lprocfs_rd_nrs,
                  .write_fptr = ptlrpc_lprocfs_wr_nrs,
                  .data       = svc},
@@ -877,4 +1145,76 @@ int lprocfs_wr_ping(struct file *file, const char *buffer,
 }
 EXPORT_SYMBOL(lprocfs_wr_ping);
 
+int nrs_orr_lprocfs_init(ptlrpc_nrs_policy_t *policy)
+{
+        struct ptlrpc_service   *svc = policy->pol_nrs->nrs_service;
+        int                      rc = 0;
+
+        struct lprocfs_vars nrs_orr_lprocfs_vars_reg[] = {
+                { .name       = "nrs_orr_quantum_reg",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_quantum,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_quantum,
+                  .data       = policy },
+                { .name       = "nrs_orr_off_type_reg",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_off_type,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_off_type,
+                  .data       = policy },
+                { .name       = "nrs_orr_trr_mode_reg",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_trr_mode,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_trr_mode,
+                  .data       = policy },
+                { .name       = "nrs_orr_supp_req_reg",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_supp_req,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_supp_req,
+                  .data       = policy },
+                { NULL }
+        };
+
+        struct lprocfs_vars nrs_orr_lprocfs_vars_hp[] = {
+                { .name       = "nrs_orr_quantum_hp",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_quantum,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_quantum,
+                  .data       = policy },
+                { .name       = "nrs_orr_off_type_hp",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_off_type,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_off_type,
+                  .data       = policy },
+                { .name       = "nrs_orr_trr_mode_hp",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_trr_mode,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_trr_mode,
+                  .data       = policy },
+                { .name       = "nrs_orr_supp_req_hp",
+                  .read_fptr  = ptlrpc_lprocfs_rd_nrs_orr_supp_req,
+                  .write_fptr = ptlrpc_lprocfs_wr_nrs_orr_supp_req,
+                  .data       = policy },
+                { NULL }
+        };
+
+        LASSERT(svc != NULL);
+
+        if (policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                rc = lprocfs_add_vars(svc->srv_procroot,
+                                      nrs_orr_lprocfs_vars_reg, NULL);
+        else
+                rc = lprocfs_add_vars(svc->srv_procroot,
+                                      nrs_orr_lprocfs_vars_hp, NULL);
+
+        RETURN(rc);
+}
+
+void nrs_orr_lprocfs_fini(ptlrpc_nrs_policy_t *policy)
+{
+        struct ptlrpc_service   *svc = policy->pol_nrs->nrs_service;
+
+        LASSERT(svc != NULL);
+
+        if (policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                lprocfs_remove_proc_entry("nrs_orr_quantum_reg",
+                                          svc->srv_procroot);
+        else
+                lprocfs_remove_proc_entry("nrs_orr_quantum_hp",
+                                          svc->srv_procroot);
+}
+
+
 #endif /* LPROCFS */
diff --git a/lustre/ptlrpc/ptlrpc_nrs.c b/lustre/ptlrpc/ptlrpc_nrs.c
index 28884dd..97f7326 100644
--- a/lustre/ptlrpc/ptlrpc_nrs.c
+++ b/lustre/ptlrpc/ptlrpc_nrs.c
@@ -43,7 +43,20 @@
 #include <obd_class.h>
 #include <lustre_net.h>
 #include <lu_object.h>
+#include <lustre/lustre_idl.h>
+#include <lustre_req_layout.h>
+#include <lustre_disk.h>
+#include <lprocfs_status.h>
 #include "ptlrpc_internal.h"
+#if 0
+#include <lustre_export.h>
+#include <lustre/lustre_idl.h>
+#include <lustre/lustre_user.h> /* memset(), memcpy() */
+#include <lustre_net.h> /* nrs_orr_req_range_t */
+#include <obd_class.h> /* obd_get_info() */
+#include <libcfs/libcfs.h>
+#include <obd.h> /* KEY_FIEMAP */
+#endif
 
 static int
 nrs_policy_init(ptlrpc_nrs_policy_t *policy, void *arg)
@@ -299,8 +312,16 @@ nrs_resource_get_safe(ptlrpc_nrs_t *nrs, ptlrpc_nrs_request_t *nrq,
         resp[NRS_RES_FALLBACK] = nrs_resource_get(fallback, nrq);
         LASSERT(resp[NRS_RES_FALLBACK] != NULL);
 
-        if (primary != NULL)
+        if (primary != NULL) {
                 resp[NRS_RES_PRIMARY] = nrs_resource_get(primary, nrq);
+                /* A primary policy may exist which may not wish to serve a
+                 * particular request for different reasons; see
+                 * nrs_orr_req_supported() for an example of this; release the
+                 * reference on the policy as it will not be used for this
+                 * request */
+                if (resp[NRS_RES_PRIMARY] == NULL)
+                        nrs_policy_put(primary);
+        }
 
         return;
 }
@@ -483,6 +504,129 @@ nrs_policy_ctl(ptlrpc_nrs_t *nrs, int type, unsigned int opc, void *arg)
                 /* XXX reserved */
                 rc = -ENOSYS;
                 break;
+
+        case PTLRPC_NRS_CTL_ORR_RD_QUANTUM: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                nrs_orr_info_t          *info = (nrs_orr_info_t *)arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                        info->oi_quantum_reg = orrd->od_quantum;
+                else if (nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_HP)
+                        info->oi_quantum_hp = orrd->od_quantum;
+                else
+                        LBUG();
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_WR_QUANTUM: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                orrd->od_quantum = *(nrs_orr_quantum_t*)arg;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_RD_OFF_TYPE: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                nrs_orr_info_t          *info = (nrs_orr_info_t *)arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                        info->oi_physical_reg = orrd->od_physical;
+                else
+                        info->oi_physical_hp = orrd->od_physical;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_WR_OFF_TYPE: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                char                    *off_type = arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (!strcmp(off_type, "physical"))
+                        orrd->od_physical = 1;
+                else if (!strcmp(off_type, "logical"))
+                        orrd->od_physical = 0;
+                else
+                        rc = -EINVAL;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_RD_TRR_MODE: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                nrs_orr_info_t          *info = (nrs_orr_info_t *)arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                        info->oi_trr_mode_reg = orrd->od_trr_mode;
+                else
+                        info->oi_trr_mode_hp = orrd->od_trr_mode;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_WR_TRR_MODE: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                char                    *trr_mode = arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (!strcmp(trr_mode, "on"))
+                        orrd->od_trr_mode = 1;
+                else if (!strcmp(trr_mode, "off"))
+                        orrd->od_trr_mode = 0;
+                else
+                        rc = -EINVAL;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_RD_SUPP_REQ: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                nrs_orr_info_t          *info = (nrs_orr_info_t *)arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG)
+                        info->oi_supp_req_reg = orrd->od_supp;
+                else
+                        info->oi_supp_req_hp = orrd->od_supp;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+                break;
+
+        case PTLRPC_NRS_CTL_ORR_WR_SUPP_REQ: {
+                struct ptlrpc_service   *svc = nrs->nrs_service;
+                nrs_orr_data_t          *orrd = policy->pol_private;
+                char                    *supp = arg;
+
+                cfs_spin_lock(&svc->srv_rq_lock);
+                if (!strcmp(supp, "read"))
+                        orrd->od_supp = NOS_OST_READ;
+                else if (!strcmp(supp, "write"))
+                        orrd->od_supp = NOS_OST_WRITE;
+                else if (!strcmp(supp, "readwrite"))
+                        orrd->od_supp = NOS_OST_RW;
+                else
+                        rc = -EINVAL;
+                cfs_spin_unlock(&svc->srv_rq_lock);
+                }
+
+                break;
+
+
         }
  out:
         if (policy != NULL)
@@ -897,12 +1041,14 @@ ptlrpc_server_nrs_setup(struct ptlrpc_service *svc)
 
         nrs = ptlrpc_server_nrs(svc, 0);
         nrs->nrs_service = svc;
+        nrs->nrs_queue_type = PTLRPC_NRS_QUEUE_REG;
         cfs_spin_lock_init(&nrs->nrs_lock);
         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_list);
         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_queued);
 
         nrs = ptlrpc_server_nrs(svc, 1);
         nrs->nrs_service = svc;
+        nrs->nrs_queue_type = PTLRPC_NRS_QUEUE_HP;
         cfs_spin_lock_init(&nrs->nrs_lock);
         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_list);
         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_queued);
@@ -1506,3 +1652,785 @@ ptlrpc_nrs_policy_desc_t ptlrpc_nrs_crr2_desc = {
         .pd_ops                 = &nrs_crr2_ops,
 };
 CFS_EXPORT_SYMBOL(ptlrpc_nrs_crr2_desc);
+
+/* Object-based Round Robin policy (ORR) */
+
+/* Comparison function for the binary heap; determines the direction of
+ * inequality between ORR requests, based on the correspoding physical or
+ * logical block numbers */
+static int
+orr_req_compare(cfs_binheap_node_t *e1, cfs_binheap_node_t *e2)
+{
+        ptlrpc_nrs_request_t *nrq1;
+        ptlrpc_nrs_request_t *nrq2;
+
+        nrq1 = container_of(e1, ptlrpc_nrs_request_t, nr_u.orr.or_node);
+        nrq2 = container_of(e2, ptlrpc_nrs_request_t, nr_u.orr.or_node);
+
+        if (nrq1->nr_u.orr.or_round < nrq2->nr_u.orr.or_round)
+                return 1;
+        if (nrq1->nr_u.orr.or_round > nrq2->nr_u.orr.or_round)
+                return 0;
+
+        /* If round numbers are equal, requests should be sorted by
+         * ascending offset */
+        if (nrq1->nr_u.orr.or_range.or_start < nrq2->nr_u.orr.or_range.or_start)
+                return 1;
+        else if (nrq1->nr_u.orr.or_range.or_start > nrq2->nr_u.orr.or_range.or_start)
+                return 0;
+        /* Requests start from the same offset */
+        else
+                /* Do the longer one first; maybe slightly more chances of
+                 * hitting the disk drive cache later with the lengthiest
+                 * request */
+                if (nrq1->nr_u.orr.or_range.or_end > nrq2->nr_u.orr.or_range.or_end)
+                        return 0;
+                else
+                        return 1;
+}
+
+/* ORR binary heap operations */
+static cfs_binheap_ops_t nrs_orr_heap_ops = {
+        .hop_enter      = NULL,
+        .hop_exit       = NULL,
+        .hop_compare    = orr_req_compare,
+};
+
+#define NRS_ORR_DFLT_OID        0x0ULL
+
+/* Populate the ORR key fields for the RPC */
+static void
+nrs_orr_key_fill(nrs_orr_data_t *orrd, ptlrpc_nrs_request_t *nrq,
+                 nrs_orr_key_t *key)
+{
+        struct ptlrpc_request  *req;
+        struct ost_body        *body;
+        struct lr_server_data  *lsd;
+        __u32                   opc;
+
+
+        req = container_of(nrq, struct ptlrpc_request, rq_nrq);
+        LASSERT(req != NULL);
+
+        opc = lustre_msg_get_opc(req->rq_reqmsg);
+
+        LASSERT(opc == OST_READ || opc == OST_WRITE);
+
+        req_capsule_init(&req->rq_pill, req, RCL_SERVER);
+
+        if (opc == OST_READ) {
+                req_capsule_set(&req->rq_pill, &RQF_OST_BRW_READ);
+                nrq->nr_u.orr.or_write = 0;
+        } else {
+                req_capsule_set(&req->rq_pill, &RQF_OST_BRW_WRITE);
+                nrq->nr_u.orr.or_write = 1;
+        }
+        if (orrd->od_trr_mode == 0) {
+                /* request pill has been initialized in ptlrpc_hpreq_init() */
+                body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
+                LASSERT(body != NULL);
+                key->ok_id = body->oa.o_id;
+        } else {
+                key->ok_id = NRS_ORR_DFLT_OID;
+        }
+
+        lsd = class_server_data(req->rq_export->exp_obd);
+        key->ok_idx = lsd->lsd_ost_index;
+}
+
+/* Is the RPC type handled by ORR ? */
+static int
+nrs_orr_req_supported(nrs_orr_data_t *orrd, ptlrpc_nrs_request_t *nrq)
+{
+        struct ptlrpc_request  *req;
+        __u32                   opc;
+
+        req = container_of(nrq, struct ptlrpc_request,rq_nrq);
+        opc = lustre_msg_get_opc(req->rq_reqmsg);
+        switch(orrd->od_supp)
+        {
+                case NOS_OST_READ:
+                        if (opc == OST_READ)
+                                RETURN(1);
+                        break;
+                case NOS_OST_WRITE:
+                        if (opc == OST_WRITE)
+                                RETURN(1);
+                        break;
+                case NOS_OST_RW:
+                        if (opc == OST_READ || opc == OST_WRITE)
+                                RETURN(1);
+                        break;
+                default:
+                        LBUG();
+        }
+        RETURN(0);
+}
+
+/* Populate the range values for the request with logical offsets */
+static void
+nrs_orr_logical(struct niobuf_remote *nb, int niocount,
+                nrs_orr_req_range_t *range)
+{
+        /* Should we do this at page boundaries ? */
+        range->or_start = nb[0].offset & CFS_PAGE_MASK;
+        range->or_end = (nb[niocount - 1].offset +
+                         nb[niocount - 1].len - 1) | ~CFS_PAGE_MASK;
+}
+
+static int
+nrs_orr_log2phys(struct obd_export *exp, struct obdo *oa,
+                 nrs_orr_req_range_t *log, int niocount,
+                 nrs_orr_req_range_t *phys);
+
+/* Populate the range values for the request with physical offsets */
+/* XXX: Maybe this should be placed in the OST layer? */
+static int
+nrs_orr_physical(struct obd_export *exp, struct obdo *oa,
+                 struct niobuf_remote *nb, int niocount,
+                 nrs_orr_req_range_t *phys)
+{
+        int                     rc;
+        nrs_orr_req_range_t     log;
+
+        /* Get logical offset range first */
+        nrs_orr_logical(nb, niocount, &log);
+
+        /* Get physical extent descriptions */
+        rc = nrs_orr_log2phys(exp, oa, &log, niocount, phys);
+
+#if 0
+        CERROR("ORR RANGES for object with ID = "LPU64" are"
+               " log start = "LPU64", log end = "LPU64" and"
+               " phys start = "LPU64", phys end = "LPU64"\n",
+               oa->o_id,log.or_start, log.or_end, phys->or_start, phys->or_end);
+#endif
+
+        RETURN(rc);
+}
+
+/* Set ORR range values in RPC */
+static void
+nrs_orr_set_range(ptlrpc_nrs_request_t *nrq, nrs_orr_req_range_t *range)
+{
+        nrq->nr_u.orr.or_range.or_start = range->or_start;
+        nrq->nr_u.orr.or_range.or_end = range->or_end;
+}
+
+#ifdef __KERNEL__
+static inline void ptlrpc_obdo_from_ostid(struct obdo *oa, struct ost_id *ostid)
+{
+        oa->o_id  = ostid_id(ostid);
+        oa->o_seq = ostid_seq(ostid);
+}
+
+static inline void ptlrpc_ioobj_from_obdo(struct obd_ioobj *ioobj, struct obdo *oa)
+{
+        ioobj->ioo_id  = oa->o_id;
+        ioobj->ioo_seq = oa->o_seq;
+}
+
+#define POSTID LPU64":"LPU64
+
+static int ptlrpc_validate_obdo(struct obd_export *exp, struct obdo *oa,
+                                struct obd_ioobj *ioobj)
+{
+        if (oa != NULL && !(oa->o_valid & OBD_MD_FLGROUP)) {
+                oa->o_seq = FID_SEQ_OST_MDT0;
+                if (ioobj)
+                        ioobj->ioo_seq = FID_SEQ_OST_MDT0;
+        /* remove fid_seq_is_rsvd() after FID-on-OST allows SEQ > 9 */
+        } else if (oa == NULL ||
+                   !(fid_seq_is_rsvd(oa->o_seq) || fid_seq_is_idif(oa->o_seq))) {
+                CERROR("%s: client %s sent invalid object "POSTID"\n",
+                       exp->exp_obd->obd_name, obd_export_nid2str(exp),
+                       oa ? oa->o_id : -1, oa ? oa->o_seq : -1);
+                return -EPROTO;
+        }
+        ptlrpc_obdo_from_ostid(oa, &oa->o_oi);
+        if (ioobj)
+                ptlrpc_ioobj_from_obdo(ioobj, oa);
+        return 0;
+}
+
+#else
+static int
+ptlrpc_validate_obdo(struct obd_export *exp, struct obdo *oa,
+                     struct obd_ioobj *ioobj)
+{
+        return 0;
+}
+#endif
+
+static int
+nrs_orr_get_range(ptlrpc_nrs_request_t *nrq, __u8 physical)
+{
+        struct ptlrpc_request  *req;
+        struct obd_ioobj       *ioo;
+        struct niobuf_remote   *nb;
+        struct ost_body        *body;
+        nrs_orr_req_range_t     range;
+        int                     niocount;
+        int                     objcount;
+        int                     rc = 0;
+        int                     i;
+
+        req = container_of(nrq, struct ptlrpc_request, rq_nrq);
+
+        objcount = req_capsule_get_size(&req->rq_pill, &RMF_OBD_IOOBJ,
+                                        RCL_CLIENT) / sizeof(*ioo);
+        /* Get and analyze fields */
+        ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
+        if (ioo == NULL)
+                RETURN(-EFAULT);
+
+        body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
+        if (body == NULL)
+                GOTO(out, rc = -EFAULT);
+
+        /* Quick hack */
+        rc = ptlrpc_validate_obdo(req->rq_export, &body->oa, ioo);
+        if (rc)
+                RETURN(rc);
+
+        /* Should this be only ioo.ioo_bufcnt? */
+        for (niocount = i = 0; i < objcount; i++)
+                niocount += ioo[i].ioo_bufcnt;
+
+        nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
+        if (nb == NULL)
+                RETURN(-EFAULT);
+
+        /* We'll run some tests with both logical and physical offset sorting of
+         * requests, and decide which ones to keep depending on the results */
+        if (physical) {
+                /* Use physical block offsets from backend filesystem extents */
+                rc = nrs_orr_physical(req->rq_export, &body->oa, nb, niocount, &range);
+                if (rc)
+                        GOTO(out, rc);
+        } else {
+                /* Use logical information from niobuf_remote structures */
+                nrs_orr_logical(nb, niocount, &range);
+        }
+        /* Assign retrieved range values to request */
+        nrs_orr_set_range(nrq, &range);
+out:
+        return rc;
+}
+
+/* Not sure if these will be ok */
+#define NRS_OBJ_HBITS_LOW       10
+#define NRS_OBJ_HBITS_HIGH      16
+#define NRS_OBJ_BBITS           8
+
+static int nrs_orr_hop_keys_eq(nrs_orr_key_t *k1, nrs_orr_key_t *k2)
+{
+        return (k1->ok_idx == k2->ok_idx && k1->ok_id == k2->ok_id);
+}
+
+/* ORR hash operations */
+static unsigned
+nrs_orr_hop_hash(cfs_hash_t *hs, const void *key, unsigned mask)
+{
+        return cfs_hash_djb2_hash(key, sizeof(nrs_orr_key_t), mask);
+}
+
+static void *
+nrs_orr_hop_key(cfs_hlist_node_t *hnode)
+{
+        nrs_orr_object_t *orro = cfs_hlist_entry(hnode, nrs_orr_object_t,
+                                                 oo_hnode);
+
+        return &orro->oo_key;
+}
+
+static int
+nrs_orr_hop_keycmp(const void *key, cfs_hlist_node_t *hnode)
+{
+        nrs_orr_object_t *orro = cfs_hlist_entry(hnode, nrs_orr_object_t,
+                                                 oo_hnode);
+
+        return nrs_orr_hop_keys_eq(&orro->oo_key, (nrs_orr_key_t *)key);
+}
+
+static void *
+nrs_orr_hop_object(cfs_hlist_node_t *hnode)
+{
+        return cfs_hlist_entry(hnode, nrs_orr_object_t, oo_hnode);
+}
+
+static void
+nrs_orr_hop_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        nrs_orr_object_t *orro = cfs_hlist_entry(hnode, nrs_orr_object_t,
+                                                 oo_hnode);
+        cfs_atomic_inc(&orro->oo_ref);
+}
+
+static void
+nrs_orr_hop_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        nrs_orr_object_t *orro = cfs_hlist_entry(hnode, nrs_orr_object_t,
+                                                 oo_hnode);
+
+        cfs_atomic_dec(&orro->oo_ref);
+}
+
+static void
+nrs_orr_hop_exit(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        nrs_orr_data_t          *orrd;
+        nrs_orr_object_t        *orro;
+
+        orro = cfs_hlist_entry(hnode, nrs_orr_object_t, oo_hnode);
+        orrd = container_of(orro->oo_res.res_parent, nrs_orr_data_t, od_res);
+
+        LASSERTF(cfs_atomic_read(&orro->oo_ref) == 0,
+                 "Busy NRS orr policy for object with objid "LPX64" at OST "
+                 "with OST index %d, with %d refs\n", orro->oo_key.ok_id,
+                 orro->oo_key.ok_idx, cfs_atomic_read(&orro->oo_ref));
+
+        OBD_SLAB_FREE_PTR(orro, orrd->od_cache);
+}
+
+static cfs_hash_ops_t nrs_orr_hash_ops = {
+        .hs_hash        = nrs_orr_hop_hash,
+        .hs_key         = nrs_orr_hop_key,
+        .hs_keycpy      = NULL,
+        .hs_keycmp      = nrs_orr_hop_keycmp,
+        .hs_object      = nrs_orr_hop_object,
+        .hs_get         = nrs_orr_hop_get,
+        .hs_put         = nrs_orr_hop_put,
+        .hs_put_locked  = nrs_orr_hop_put,
+        .hs_exit        = nrs_orr_hop_exit
+};
+
+/* XXX: Temporarily here */
+#ifdef __KERNEL__
+extern int nrs_orr_lprocfs_init(ptlrpc_nrs_policy_t *policy);
+extern void nrs_orr_lprocfs_fini(ptlrpc_nrs_policy_t *policy);
+#else
+int nrs_orr_lprocfs_init(ptlrpc_nrs_policy_t *policy) { return 0; }
+void nrs_orr_lprocfs_fini(ptlrpc_nrs_policy_t *policy) {}
+#endif
+
+#define NRS_ORR_QUANTUM_DFLT 8
+/* Offset ordering type, 0 for logical offsets, 1 for physical */
+#define NRS_ORR_OFFSET_TYPE 1
+
+static int
+nrs_orr_start(ptlrpc_nrs_policy_t *policy)
+{
+        nrs_orr_data_t         *orrd;
+        int                     rc = 0;
+
+        ENTRY;
+
+        OBD_ALLOC_PTR(orrd);
+        if (orrd == NULL)
+                RETURN(-ENOMEM);
+
+        /* Binary heap for sorted incoming requests */
+        orrd->od_binheap = cfs_binheap_create(&nrs_orr_heap_ops,
+                                              CBH_FLAG_ATOMIC_GROW,
+                                              4096, NULL);
+        if (orrd->od_binheap == NULL)
+                GOTO(failed, rc = -ENOMEM);
+
+        /* Slab cache for NRS ORR objects */
+        orrd->od_cache = cfs_mem_cache_create(policy->pol_nrs->nrs_queue_type ==
+                                              PTLRPC_NRS_QUEUE_REG ?
+                                              "nrs_orr_obj_reg" :
+                                              "nrs_orr_obj_hp",
+                                              sizeof(nrs_orr_object_t),
+                                              0, 0);
+        if (orrd->od_cache == NULL)
+                GOTO(failed, rc = -ENOMEM);
+
+        /* Hash for finding objects by nrs_orr_key_t */
+        orrd->od_obj_hash = cfs_hash_create(policy->pol_nrs->nrs_queue_type ==
+                                            PTLRPC_NRS_QUEUE_REG ?
+                                            "nrs_obj_hash_reg" :
+                                            "nrs_obj_hash_hp",
+                                            NRS_OBJ_HBITS_LOW,
+                                            NRS_OBJ_HBITS_HIGH,
+                                            NRS_OBJ_BBITS, 0,
+                                            CFS_HASH_MIN_THETA,
+                                            CFS_HASH_MAX_THETA,
+                                            &nrs_orr_hash_ops,
+                                            CFS_HASH_DEFAULT);
+
+        if (orrd->od_obj_hash == NULL)
+                GOTO(failed, rc = -ENOMEM);
+
+        orrd->od_quantum = NRS_ORR_QUANTUM_DFLT;
+        orrd->od_supp = NOS_DFLT;
+        orrd->od_physical = NRS_ORR_OFFSET_TYPE;
+
+        rc = nrs_orr_lprocfs_init(policy);
+        if (rc != 0)
+                GOTO(failed, rc);
+
+        policy->pol_private = orrd;
+
+        RETURN(rc);
+
+failed:
+        if (orrd->od_obj_hash) {
+                cfs_hash_putref(orrd->od_obj_hash);
+                orrd->od_obj_hash = NULL;
+        }
+        if (orrd->od_cache) {
+                rc = cfs_mem_cache_destroy(orrd->od_cache);
+                LASSERTF(rc == 0, "could not destroy od_cache slab\n");
+        }
+        if (orrd->od_binheap != NULL)
+                cfs_binheap_destroy(orrd->od_binheap);
+
+        OBD_FREE_PTR(orrd);
+
+        RETURN(rc);
+}
+
+static void
+nrs_orr_stop(struct ptlrpc_nrs_policy *policy)
+{
+        nrs_orr_data_t *orrd = policy->pol_private;
+
+        ENTRY;
+
+        LASSERT(orrd != NULL);
+        LASSERT(cfs_binheap_is_empty(orrd->od_binheap));
+
+        cfs_binheap_destroy(orrd->od_binheap);
+        cfs_hash_putref(orrd->od_obj_hash);
+        cfs_mem_cache_destroy(orrd->od_cache);
+
+        nrs_orr_lprocfs_fini(policy);
+
+        OBD_FREE_PTR(orrd);
+
+        policy->pol_private = NULL;
+}
+
+static int
+nrs_orr_res_get(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_request_t *nrq,
+                ptlrpc_nrs_resource_t *parent, ptlrpc_nrs_resource_t **resp)
+{
+        nrs_orr_data_t         *orrd;
+        nrs_orr_object_t       *orro;
+        nrs_orr_object_t       *tmp;
+        nrs_orr_key_t           key;
+
+        ENTRY;
+
+        /* nrs_orr_data_t is requested */
+        if (parent == NULL) {
+                *resp = &((nrs_orr_data_t*)policy->pol_private)->od_res;
+                RETURN(0);
+        }
+
+        orrd = container_of(parent, nrs_orr_data_t, od_res);
+        /* If the request type is not supported, fail the enqueuing; the RPC
+         * will be handled by the default NRS policy. */
+        if (!nrs_orr_req_supported(orrd, nrq))
+                RETURN(-1);
+
+        /* nrs_orr_object_t is requested */
+        nrs_orr_key_fill(orrd, nrq, &key);
+
+        orro = cfs_hash_lookup(orrd->od_obj_hash, &key);
+        if (orro != NULL)
+                GOTO(out, 1);
+
+        OBD_SLAB_ALLOC_PTR_GFP(orro, orrd->od_cache, CFS_ALLOC_IO);
+        if (orro == NULL)
+                RETURN(-ENOMEM);
+
+        orro->oo_key = key;
+        /* This needs to be locked, really, but no harm like this */
+        orro->oo_quantum = orrd->od_quantum;
+
+        cfs_atomic_set(&orro->oo_ref, 1);
+        tmp = cfs_hash_findadd_unique(orrd->od_obj_hash, &orro->oo_key,
+                                      &orro->oo_hnode);
+        if (tmp != orro) {
+                /* TODO: I think this should never happen as we are under a lock
+                 * here; recheck */
+                OBD_SLAB_FREE_PTR(orro, orrd->od_cache);
+                orro = tmp;
+        }
+out:
+        /* XXX: For debugging purposes */
+        nrq->nr_u.orr.or_key = orro->oo_key;
+
+        *resp = &orro->oo_res;
+
+        RETURN(1);
+}
+
+static void
+nrs_orr_res_put(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_resource_t *res)
+{
+        nrs_orr_data_t          *orrd;
+        nrs_orr_object_t        *orro;
+
+        ENTRY;
+
+        if (res->res_parent == NULL)
+                return;
+
+        orro = container_of(res, nrs_orr_object_t, oo_res);
+        LASSERT(res->res_parent != NULL);
+        orrd = container_of(res->res_parent, nrs_orr_data_t, od_res);
+
+        cfs_hash_put(orrd->od_obj_hash, &orro->oo_hnode);
+        /* XXX: Maybe we want to garbadge-collect or LRU for ORR objects,
+         * as they will accumulate after a while, and if the ORR policy does
+         * not get stopped they will remain in memory.
+         */
+}
+
+/* Picks up a request from the root of the binheap */
+static ptlrpc_nrs_request_t *
+nrs_orr_req_poll(ptlrpc_nrs_policy_t *policy, void *arg)
+{
+        nrs_orr_data_t          *orrd = policy->pol_private;
+        cfs_binheap_node_t      *node = cfs_binheap_root(orrd->od_binheap);
+
+        return node == NULL ? NULL :
+                container_of(node, ptlrpc_nrs_request_t, nr_u.orr.or_node);
+}
+
+/* Sort-adds a request to the binary heap */
+static int
+nrs_orr_req_add(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_request_t *nrq)
+{
+        nrs_orr_data_t          *orrd;
+        nrs_orr_object_t        *orro;
+        int                      rc;
+        struct ptlrpc_request   *req = container_of(nrq, struct ptlrpc_request,
+                                                    rq_nrq);
+        struct ptlrpc_service   *svc = req->rq_rqbd->rqbd_service;
+
+        orro = container_of(nrs_request_resource(nrq),
+                            nrs_orr_object_t, oo_res);
+        orrd = container_of(nrs_request_resource(nrq)->res_parent,
+                            nrs_orr_data_t, od_res);
+
+        LASSERT(orro->oo_res.res_parent != NULL);
+        LASSERT(orro->oo_res.res_parent == nrs_request_resource(nrq)->res_parent);
+
+        /* Release the lock here temporarily (we don't really need it here), as
+         * some operations that need to be carried out as part of performing the
+         * fimeap call may need to sleep.
+         */
+        cfs_spin_unlock(&svc->srv_rq_lock);
+        rc = nrs_orr_get_range(nrq, orrd->od_physical == 1 &&
+                               nrq->nr_u.orr.or_write == 0);
+        cfs_spin_lock(&svc->srv_rq_lock);
+        if (rc)
+                RETURN(rc);
+
+        if (orro->oo_started == 0) {
+                orro->oo_round = orrd->od_round;
+                orrd->od_round++;
+                orro->oo_started = 1;
+        }
+
+        nrq->nr_u.orr.or_round = orro->oo_round;
+        rc = cfs_binheap_insert(orrd->od_binheap, &nrq->nr_u.orr.or_node);
+        if (rc == 0) {
+                orro->oo_quantum--;
+                if (orro->oo_quantum == 0) {
+                        orro->oo_quantum = orrd->od_quantum;
+                        orro->oo_started = 0;
+                }
+        }
+        RETURN(rc);
+}
+
+static void
+nrs_orr_req_del(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_request_t *nrq)
+{
+        nrs_orr_data_t          *orrd;
+        nrs_orr_object_t        *orro;
+
+        orrd = container_of(nrs_request_resource(nrq)->res_parent,
+                            nrs_orr_data_t, od_res);
+        orro = container_of(nrs_request_resource(nrq),
+                            nrs_orr_object_t, oo_res);
+
+        LASSERT(nrq->nr_u.orr.or_round < orrd->od_round);
+        LASSERT(nrq->nr_u.orr.or_round <= orro->oo_round);
+
+        cfs_binheap_remove(orrd->od_binheap, &nrq->nr_u.orr.or_node);
+}
+
+static void
+nrs_orr_req_start(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_request_t *nrq)
+{
+        nrs_orr_object_t        *orro = container_of(nrs_request_resource(nrq),
+                                                     nrs_orr_object_t, oo_res);
+
+        LASSERT(nrq->nr_u.orr.or_round <= orro->oo_round);
+
+        /* NB: resource control, credits etc can be added to here */
+        CDEBUG(D_RPCTRACE, "NRS start %s request for object with ID "LPU64" "
+               "from OST with index %d, with round "LPU64"\n",
+               nrs_request_policy(nrq)->pol_name, nrq->nr_u.orr.or_key.ok_id,
+               nrq->nr_u.orr.or_key.ok_idx, nrq->nr_u.orr.or_round);
+}
+
+static void
+nrs_orr_req_stop(ptlrpc_nrs_policy_t *policy, ptlrpc_nrs_request_t *nrq)
+{
+        nrs_orr_object_t        *orro = container_of(nrs_request_resource(nrq),
+                                                     nrs_orr_object_t, oo_res);
+
+        LASSERT(nrq->nr_u.orr.or_round <= orro->oo_round);
+
+        /* NB: resource control, credits etc can be added to here */
+        CDEBUG(D_RPCTRACE, "NRS stop %s request for object with ID "LPU64" "
+               "from OST with index %d, with round "LPU64"\n",
+               nrs_request_policy(nrq)->pol_name, nrq->nr_u.orr.or_key.ok_id,
+               nrq->nr_u.orr.or_key.ok_idx, nrq->nr_u.orr.or_round);
+}
+
+static ptlrpc_nrs_ops_t nrs_orr_ops = {
+        .op_policy_init         = NULL,
+        .op_policy_fini         = NULL,
+        .op_policy_start        = nrs_orr_start,
+        .op_policy_stop         = nrs_orr_stop,
+        .op_policy_ctl          = NULL,
+        .op_res_get             = nrs_orr_res_get,
+        .op_res_put             = nrs_orr_res_put,
+        .op_req_poll            = nrs_orr_req_poll,
+        .op_req_enqueue         = nrs_orr_req_add,
+        .op_req_dequeue         = nrs_orr_req_del,
+        .op_req_start           = nrs_orr_req_start,
+        .op_req_stop            = nrs_orr_req_stop
+};
+
+ptlrpc_nrs_policy_desc_t ptlrpc_nrs_orr_desc = {
+        .pd_name                = "orr",
+        .pd_type                = PTLRPC_NRS_TYPE_ORR,
+        .pd_ops                 = &nrs_orr_ops
+};
+CFS_EXPORT_SYMBOL(ptlrpc_nrs_orr_desc);
+
+#ifdef __KERNEL__
+
+#define ORR_NUM_EXTENTS 1
+
+/* This structure mirrors struct ll_user_fiemap, but uses a fixed length for the
+ * array of ll_fiemap_extent structs; its prime intention is for sttaically
+ * allocated variables.
+ */
+typedef struct nrs_orr_fiemap {
+        __u64 of_start;
+        __u64 of_length;
+        __u32 of_flags;
+        __u32 of_mapped_extents;
+        __u32 of_extent_count;
+        __u32 of_reserved;
+        struct ll_fiemap_extent of_extents[ORR_NUM_EXTENTS];
+} nrs_orr_fiemap_t;
+
+/* Get extents from the range of niobuf_remotes here by doing fiemap calls via
+ * obd_get_info(), and then return a lower and higher physical block number for
+ * the RPC. This would most probably need another level of translation when OSTs
+ * are created on top of RAID arrays, in order to obtain the actual physical
+ * block numbers on the disk drives; what is returned from ldiskfs extents in
+ * such cases must be logical RAID block numbers, but maybe we can safely ignore
+ * this; hopefully mdraid will assign logical block numbers to each disk in turn;
+ */
+static int
+nrs_orr_log2phys(struct obd_export *exp, struct obdo *oa,
+                 nrs_orr_req_range_t *log, int niocount,
+                 nrs_orr_req_range_t *phys)
+{
+        /* Use ll_fiemap_info_key, as done in ost_get_info() */
+        struct ll_fiemap_info_key  fm_key = { .name = KEY_FIEMAP };
+        nrs_orr_fiemap_t           of_fiemap;
+        struct ll_user_fiemap     *fiemap = (struct ll_user_fiemap *)&of_fiemap;
+        //int                        i;
+        int                        rc;
+        int                        replylen;
+        loff_t                     start = OBD_OBJECT_EOF;
+        loff_t                     end = 0;
+        //__u64                      log_ext_start = log->or_start;
+        __u64                      log_req_len = log->or_end - log->or_start;
+
+        ENTRY;
+
+        fiemap = (struct ll_user_fiemap *)&of_fiemap;
+
+        fm_key.oa = *oa;
+        fm_key.fiemap.fm_start = log->or_start;
+        fm_key.fiemap.fm_length = log_req_len;
+        fm_key.fiemap.fm_extent_count = ORR_NUM_EXTENTS;
+
+        /* Here we perform a fiemap call in order to get extent descriptions;
+         * ideally, we could pass '0' in to fm_extent_count in order to obtain
+         * the number of extents required in order to map the whole file and
+         * allocate as much memory as we require, but here we adopt a faster
+         * route by performing only one fiemap call to get the extent
+         * information and assume the nrs_orr_fiemap_t struct that is allocated
+         * on the stack will suffice; i fit does not, we can either fail the
+         * operation, or resort to allocating memory on demand or from a pool.
+         */
+        /* Do an obd_get_info to get the extent descriptions */
+        rc = obd_get_info(exp, sizeof(fm_key), &fm_key, &replylen,
+                          (struct ll_user_fiemap *)&of_fiemap, NULL);
+        if (unlikely(rc)) {
+                CERROR("obd_get_info failed: rc = %d\n", rc);
+                goto out;
+        }
+
+        if (unlikely(fiemap->fm_mapped_extents == 0 ||
+            fiemap->fm_mapped_extents > ORR_NUM_EXTENTS)) {
+                CERROR("fm_mapped_extents is %u.\n", fiemap->fm_mapped_extents);
+                GOTO(out, rc = -EFAULT);
+        }
+
+#if 0
+        /* Return extent information to caller */
+        for (i = 0; i < fiemap->fm_mapped_extents; i++)
+        {
+                if (likely(start > fiemap->fm_extents[i].fe_physical)) {
+                        start = fiemap->fm_extents[i].fe_physical;
+                        log_ext_start = fiemap->fm_extents[i].fe_logical;
+                }
+                if (likely(end < fiemap->fm_extents[i].fe_physical +
+                                fiemap->fm_extents[i].fe_length))
+                        end = fiemap->fm_extents[i].fe_physical +
+                                fiemap->fm_extents[i].fe_length;
+        }
+        start = start + (log->or_start - log_ext_start);
+        LASSERTF(start < end, "start = %llx, end = %llx\n", start, end);
+#endif
+        /* Optimize start and end calculation for the one fiemap case; this will
+         * have to be changed if we include more than one fiemap structs in the
+         * future.
+         */
+        start = fiemap->fm_extents[0].fe_physical;
+        start += log->or_start - fiemap->fm_extents[0].fe_logical;
+        end = start + log_req_len;
+
+        phys->or_start = start;
+        phys->or_end = end;
+out:
+
+        return rc;
+}
+#else
+/* Temporary physical offset stub for liblustre */
+static int
+nrs_orr_log2phys(struct obd_export *exp, struct obdo *oa,
+                 nrs_orr_req_range_t *log, int niocount,
+                 nrs_orr_req_range_t *phys)
+{
+        return 0;
+}
+#endif
diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index cbdf422..d9acd50 100644
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -588,6 +588,13 @@ ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size, int max_reply_size,
         if (rc != 0)
                 GOTO(failed, NULL);
 
+        if (!strcmp(name, "ost_io")) {
+                rc = ptlrpc_nrs_policy_register(service, PTLRPC_NRS_QUEUE_BOTH,
+                                                &ptlrpc_nrs_orr_desc, 0, NULL);
+        if (rc != 0)
+                GOTO(failed, NULL);
+        }
+
         cfs_spin_lock (&ptlrpc_all_services_lock);
         cfs_list_add (&service->srv_list, &ptlrpc_all_services);
         cfs_spin_unlock (&ptlrpc_all_services_lock);
@@ -2611,8 +2618,6 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
         cfs_list_del_init (&service->srv_list);
         cfs_spin_unlock (&ptlrpc_all_services_lock);
 
-        ptlrpc_lprocfs_unregister_service(service);
-
         /* All history will be culled when the next request buffer is
          * freed */
         service->srv_max_history_rqbds = 0;
@@ -2697,6 +2702,9 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
 
         ptlrpc_server_nrs_cleanup(service);
 
+        /* XXX: Moving this here may potentially introduce other issues */
+        ptlrpc_lprocfs_unregister_service(service);
+
         /* Now free all the request buffers since nothing references them
          * any more... */
         while (!cfs_list_empty(&service->srv_idle_rqbds)) {
