Branch HEAD

b=18478
i=jay
i=liang

a new osc is added for echo client purpose (instead of re-using an existing one)
a test is added

diff --git a/lustre-iokit/obdfilter-survey/libecho b/lustre-iokit/obdfilter-survey/libecho
index 1ae1dc0..de52762 100644 (file)
--- a/lustre-iokit/obdfilter-survey/libecho
+++ b/lustre-iokit/obdfilter-survey/libecho
@@ -176,7 +176,7 @@ get_devnos () {
 # do cleanup for netdisk case.
 cleanup_netdisk () {
     for osc in $@; do
 # do cleanup for netdisk case.
 cleanup_netdisk () {
     for osc in $@; do

-        lctl <<EOF
+        $lctl <<EOF

         cfg_device $osc 
         cleanup
         detach
         cfg_device $osc 
         cleanup
         detach


@@ -187,18 +187,18 @@ EOF
 # do cleanup for network case.
 cleanup_network () {
     local clean_srv_OSS=$1
 # do cleanup for network case.
 cleanup_network () {
     local clean_srv_OSS=$1

-    lctl <<EOF
+    $lctl <<EOF

     cfg_device echotmp 
     cleanup
     detach
 EOF
     cfg_device echotmp 
     cleanup
     detach
 EOF

-    remote_shell "root@$server_nid" "lctl << EOF
+    remote_shell "root@$server_nid" "$lctl << EOF

         cfg_device echo_srv
         cleanup
         detach
 EOF"
     if [ $clean_srv_OSS ]; then
         cfg_device echo_srv
         cleanup
         detach
 EOF"
     if [ $clean_srv_OSS ]; then

-        remote_shell "root@$server_nid" "lctl << EOF
+        remote_shell "root@$server_nid" "$lctl << EOF

             cfg_device OSS
             cleanup
             detach
             cfg_device OSS
             cleanup
             detach


@@ -248,7 +248,7 @@ cleanup () {
 }
 trap cleanup SIGHUP SIGINT SIGTERM
 
 }
 trap cleanup SIGHUP SIGINT SIGTERM
 

-# gets echoclient device number and attch it to the client UUID
+# gets echoclient device number and attach it to the client UUID

 # 
 # parameter: 1. hostname
 #           2. client name, ex:- ns8:ECHO_ns8
 # 
 # parameter: 1. hostname
 #           2. client name, ex:- ns8:ECHO_ns8


@@ -384,7 +384,7 @@ split_hostname () {
 
 check_cleanup () {
     type_obj="$1"
 
 check_cleanup () {
     type_obj="$1"

-    osc_names_str=$(lctl dl | grep $type_obj)
+    osc_names_str=$($lctl dl | grep $type_obj)

     count=0;
     for name in $osc_names_str; do
         count=$((count+1))
     count=0;
     for name in $osc_names_str; do
         count=$((count+1))


@@ -399,7 +399,7 @@ check_cleanup () {
 
 check_setup () {
     type_obj="$1"
 
 check_setup () {
     type_obj="$1"

-    osc_names_str=$(lctl dl | grep $type_obj)
+    osc_names_str=$($lctl dl | grep $type_obj)

     count=0;
     for name in $osc_names_str; do
         count=$((count+1))
     count=0;
     for name in $osc_names_str; do
         count=$((count+1))

diff --git a/lustre-iokit/obdfilter-survey/obdfilter-survey b/lustre-iokit/obdfilter-survey/obdfilter-survey
index 7b83a8e..7ce397f 100755 (executable)
--- a/lustre-iokit/obdfilter-survey/obdfilter-survey
+++ b/lustre-iokit/obdfilter-survey/obdfilter-survey
@@ -269,32 +269,41 @@ for trgt in $targets; do
     ndevs=$((ndevs+1))
 done
 if [ $case == "netdisk" ]; then
     ndevs=$((ndevs+1))
 done
 if [ $case == "netdisk" ]; then

-    if [ "$targets" ]; then
-        for ((i = 0; i < $ndevs; i++)); do
-             setup_osc_for_remote_ost ${host_names[$i]} ${client_names[$i]} $i
-            cleanup_oscs="$cleanup_oscs ${client_names[$i]}_osc"
-            host_names[$i]=localhost
-        done
-    fi
-        declare -a osc_names
-        declare -a osc_uuids
-        osc_names_str=$(lctl dl |grep osc | awk "{if (\$2 == \"UP\" && \$3 == \"osc\") {print \$4} }")
-        count=0;
-        for name in $osc_names_str; do
-            osc_names[$count]=$name
-            count=$((count+1))
-        done
-        osc_uuid_str=$(lctl dl |grep osc | awk "{if (\$2 == \"UP\" && \$3 == \"osc\") {print \$5} }")
-        count=0;
-        for uuid in $osc_uuid_str; do
-            osc_uuids[$count]=$uuid
-            count=$((count+1))
-        done
-        for (( i = 0; i < $count; i++ ))
-        do
-            ec_using_osc ${osc_names[$i]}
-        done
-        echo_clients=$(lctl dl | grep echo_client | awk "{if (\$2 == \"UP\" && \$3 == \"echo_client\") {print \$4} }")
+        if [ "$targets" ]; then
+            for ((i = 0; i < $ndevs; i++)); do
+                setup_osc_for_remote_ost ${host_names[$i]} \
+                                         ${client_names[$i]} $i
+                osc_name=${client_names[$i]}_osc
+                ec_using_osc $osc_name
+               cleanup_oscs="$cleanup_oscs $osc_name"
+            done
+        else
+            client_names_str=$($lctl dl | grep -v mdt | \
+                awk '{if ($2 == "UP" && $3 == "osc") {print $4} }')
+            count=0;
+            for name in $client_names_str; do
+                client_names[$count]=`echo $name | sed 's/-osc-.*$//'`
+                count=$((count+1))
+            done
+
+            host_names_str=$($lctl dl -t | grep -v mdt | \
+                awk '{if ($2 == "UP" && $3 == "osc") {print $7} }')
+            count=0;
+            for name in $host_names_str; do
+                host_names[$count]=`echo $name | sed 's/@.*$//'`
+                count=$((count+1))
+            done
+
+            for (( i = 0; i < $count; i++ )) do
+                setup_osc_for_remote_ost ${host_names[$i]} \
+                                         ${client_names[$i]} $i
+                osc_name=${client_names[$i]}_osc
+                ec_using_osc $osc_name
+               cleanup_oscs="$cleanup_oscs $osc_name"
+            done
+        fi
+
+        echo_clients=$($lctl dl | grep echo_client | awk "{if (\$2 == \"UP\" && \$3 == \"echo_client\") {print \$4} }")

         cnt=0;
         for name in $echo_clients; do
             client_names[$cnt]=$name
         cnt=0;
         for name in $echo_clients; do
             client_names[$cnt]=$name


@@ -315,21 +324,21 @@ if [ $case == "network" ]; then
     fi
     # Now do the server setup
     setup_srv_obd $server_nid "echo_srv"
     fi
     # Now do the server setup
     setup_srv_obd $server_nid "echo_srv"

-    oss_on_srv=`dsh $server_nid root "lctl dl | grep OSS" | awk '{ print $4 }'`
+    oss_on_srv=`dsh $server_nid root "$lctl dl | grep OSS" | awk '{ print $4 }'`

     if [ -z $oss_on_srv ]; then
         setup_OSS $server_nid
         clean_srv_OSS=1
     fi
     if [ -z $oss_on_srv ]; then
         setup_OSS $server_nid
         clean_srv_OSS=1
     fi

-    if ! dsh $server_nid root "lctl dl | grep obdecho > /dev/null 2>&1"; then
+    if ! dsh $server_nid root "$lctl dl | grep obdecho > /dev/null 2>&1"; then

         echo "obdecho not setup on server"
         exit 1
     fi
         echo "obdecho not setup on server"
         exit 1
     fi

-    if ! dsh $server_nid root "lctl dl | grep ost > /dev/null 2>&1"; then
+    if ! dsh $server_nid root "$lctl dl | grep ost > /dev/null 2>&1"; then

         echo "ost not setup on server"
         exit 1
     fi
     # Now start client setup
         echo "ost not setup on server"
         exit 1
     fi
     # Now start client setup

-    osc_names_str=$(lctl dl)
+    osc_names_str=$($lctl dl| grep osc | grep -v mdt | grep UP)

     if [ -n "$osc_names_str" ]; then
         echo "The existing setup must be cleaned";
         exit 0;
     if [ -n "$osc_names_str" ]; then
         echo "The existing setup must be cleaned";
         exit 0;


@@ -409,7 +418,7 @@ for ((rsz = $rszlo; rsz <= $rszhi; rsz*=2)); do
            actual_size=$((actual_rsz*count*thr))
             total_size=$((actual_size*ndevs))
            # show computed parameters
            actual_size=$((actual_rsz*count*thr))
             total_size=$((actual_size*ndevs))
            # show computed parameters

-           str=`printf 'ost %2d sz %8dK rsz %4d obj %4d thr %4d ' \
+           str=`printf 'ost %2d sz %8dK rsz %4dK obj %4d thr %4d ' \

                     $ndevs $total_size $actual_rsz $total_nobj $total_thr`
            echo "=======================> $str" >> $workf
            print_summary -n "$str"
                     $ndevs $total_size $actual_rsz $total_nobj $total_thr`
            echo "=======================> $str" >> $workf
            print_summary -n "$str"


@@ -480,6 +489,7 @@ for ((rsz = $rszlo; rsz <= $rszhi; rsz*=2)); do
                 for host in ${unique_hosts[@]}; do
                     rm ${cmdsf}_${host}
                 done
                 for host in ${unique_hosts[@]}; do
                     rm ${cmdsf}_${host}
                 done

+

                # compute bandwidth from total data / elapsed time
                str=`awk "BEGIN {printf \"%7.2f \",\
                         $total_size / (( $t1 - $t0 ) * 1024)}"`
                # compute bandwidth from total data / elapsed time
                str=`awk "BEGIN {printf \"%7.2f \",\
                         $total_size / (( $t1 - $t0 ) * 1024)}"`

diff --git a/lustre/obdclass/lprocfs_status.c b/lustre/obdclass/lprocfs_status.c
index fbcf034..fb23209 100644 (file)
--- a/lustre/obdclass/lprocfs_status.c
+++ b/lustre/obdclass/lprocfs_status.c
@@ -634,9 +634,8 @@ int lprocfs_rd_conn_uuid(char *page, char **start, off_t off, int count,
 
         LPROCFS_CLIMP_CHECK(obd);
         conn = obd->u.cli.cl_import->imp_connection;
 
         LPROCFS_CLIMP_CHECK(obd);
         conn = obd->u.cli.cl_import->imp_connection;

-        LASSERT(conn != NULL);

         *eof = 1;
         *eof = 1;

-        if (obd->u.cli.cl_import) {
+        if (conn && obd->u.cli.cl_import) {

                 rc = snprintf(page, count, "%s\n",
                               conn->c_remote_uuid.uuid);
         } else {
                 rc = snprintf(page, count, "%s\n",
                               conn->c_remote_uuid.uuid);
         } else {

diff --git a/lustre/obdecho/echo_client.c b/lustre/obdecho/echo_client.c
index 3fcfdbc..504196a 100644 (file)
--- a/lustre/obdecho/echo_client.c
+++ b/lustre/obdecho/echo_client.c
@@ -90,6 +90,7 @@ struct echo_lock {
         struct list_head       el_chain;
         struct echo_object    *el_object;
         __u64                  el_cookie;
         struct list_head       el_chain;
         struct echo_object    *el_object;
         __u64                  el_cookie;

+        atomic_t               el_refcount;

 };
 
 struct echo_io {
 };
 
 struct echo_io {


@@ -408,6 +409,7 @@ static int echo_lock_init(const struct lu_env *env,
                 cl_lock_slice_add(lock, &el->el_cl, obj, &echo_lock_ops);
                 el->el_object = cl2echo_obj(obj);
                 CFS_INIT_LIST_HEAD(&el->el_chain);
                 cl_lock_slice_add(lock, &el->el_cl, obj, &echo_lock_ops);
                 el->el_object = cl2echo_obj(obj);
                 CFS_INIT_LIST_HEAD(&el->el_chain);

+                atomic_set(&el->el_refcount, 0);

         }
         RETURN(el == NULL ? -ENOMEM : 0);
 }
         }
         RETURN(el == NULL ? -ENOMEM : 0);
 }


@@ -763,6 +765,24 @@ static struct lu_device *echo_device_fini(const struct lu_env *env,
         return NULL;
 }
 
         return NULL;
 }
 

+static void echo_lock_release(const struct lu_env *env,
+                              struct echo_lock *ecl,
+                              int still_used)
+{
+        struct cl_lock *clk = echo_lock2cl(ecl);
+
+        cl_lock_get(clk);
+        cl_unuse(env, clk);
+        cl_lock_release(env, clk, "ec enqueue", ecl->el_object);
+        if (!still_used) {
+                cl_lock_mutex_get(env, clk);
+                cl_lock_cancel(env, clk);
+                cl_lock_delete(env, clk);
+                cl_lock_mutex_put(env, clk);
+        }
+        cl_lock_put(env, clk);
+}
+

 static struct lu_device *echo_device_free(const struct lu_env *env,
                                           struct lu_device *d)
 {
 static struct lu_device *echo_device_free(const struct lu_env *env,
                                           struct lu_device *d)
 {


@@ -778,23 +798,18 @@ static struct lu_device *echo_device_free(const struct lu_env *env,
         while (!list_empty(&ec->ec_locks)) {
                 struct echo_lock *ecl = list_entry(ec->ec_locks.next,
                                                    struct echo_lock, el_chain);
         while (!list_empty(&ec->ec_locks)) {
                 struct echo_lock *ecl = list_entry(ec->ec_locks.next,
                                                    struct echo_lock, el_chain);

-                struct cl_lock *lock  = echo_lock2cl(ecl);
+                int still_used = 0;

 
 

-                list_del_init(&ecl->el_chain);
+                if (atomic_dec_and_test(&ecl->el_refcount))
+                        list_del_init(&ecl->el_chain);
+                else
+                        still_used = 1;

                 spin_unlock(&ec->ec_lock);
 
                 spin_unlock(&ec->ec_lock);
 

-                CERROR("echo client: pending lock %p\n", ecl);
-
-                cl_lock_get(lock);
-                cl_unuse(env, lock);
-                cl_lock_release(env, lock, "ec enqueue", ecl->el_object);
-
-                cl_lock_mutex_get(env, lock);
-                cl_lock_cancel(env, lock);
-                cl_lock_delete(env, lock);
-                cl_lock_mutex_put(env, lock);
-                cl_lock_put(env, lock);
+                CERROR("echo client: pending lock %p refs %d\n",
+                       ecl, atomic_read(&ecl->el_refcount));

 
 

+                echo_lock_release(env, ecl, still_used);

                 spin_lock(&ec->ec_lock);
         }
         spin_unlock(&ec->ec_lock);
                 spin_lock(&ec->ec_lock);
         }
         spin_unlock(&ec->ec_lock);


@@ -959,53 +974,75 @@ static int cl_echo_object_put(struct echo_object *eco)
         RETURN(0);
 }
 
         RETURN(0);
 }
 

-static int cl_echo_enqueue(struct echo_object *eco, obd_off start, obd_off end,
-                           int mode, __u64 *cookie)
+static int cl_echo_enqueue0(struct lu_env *env, struct echo_object *eco,
+                            obd_off start, obd_off end, int mode,
+                            __u64 *cookie , __u32 enqflags)

 {
 {

-        struct lu_env *env;
-        struct cl_lock *lck;
-        struct echo_thread_info *info;

         struct cl_io *io;
         struct cl_io *io;

+        struct cl_lock *lck;
+        struct cl_object *obj;

         struct cl_lock_descr *descr;
         struct cl_lock_descr *descr;

-        struct cl_object *obj = echo_obj2cl(eco);
-        int refcheck;
-        int result;
+        struct echo_thread_info *info;
+        int rc = -ENOMEM;

         ENTRY;
 
         ENTRY;
 

-        env = cl_env_get(&refcheck);
-        if (IS_ERR(env))
-                RETURN(PTR_ERR(env));
-

         info = echo_env_info(env);
         info = echo_env_info(env);

+        io = &info->eti_io;

         descr = &info->eti_descr;
         descr = &info->eti_descr;

+        obj = echo_obj2cl(eco);
+

         descr->cld_obj   = obj;
         descr->cld_start = cl_index(obj, start);
         descr->cld_end   = cl_index(obj, end);
         descr->cld_mode  = mode == LCK_PW ? CLM_WRITE : CLM_READ;
         descr->cld_obj   = obj;
         descr->cld_start = cl_index(obj, start);
         descr->cld_end   = cl_index(obj, end);
         descr->cld_mode  = mode == LCK_PW ? CLM_WRITE : CLM_READ;

-
-        io = &info->eti_io;

         io->ci_obj = obj;
         io->ci_obj = obj;

-        result = cl_io_init(env, io, CIT_MISC, obj);
-        if (result < 0)
-                GOTO(out, result);
-        LASSERT(result == 0);

 
 

-        result = -ENOMEM;
-        lck = cl_lock_request(env, io, descr, CEF_ASYNC, "ec enqueue", eco);
+        lck = cl_lock_request(env, io, descr, CEF_ASYNC | enqflags,
+                              "ec enqueue", eco);

         if (lck) {
                 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
                 struct echo_lock *el;
 
         if (lck) {
                 struct echo_client_obd *ec = eco->eo_dev->ed_ec;
                 struct echo_lock *el;
 

-                result = cl_wait(env, lck);
-                if (result == 0) {
+                rc = cl_wait(env, lck);
+                if (rc == 0) {

                         el = cl2echo_lock(cl_lock_at(lck, &echo_device_type));
                         spin_lock(&ec->ec_lock);
                         el = cl2echo_lock(cl_lock_at(lck, &echo_device_type));
                         spin_lock(&ec->ec_lock);

-                        list_add(&el->el_chain, &ec->ec_locks);
-                        *cookie = el->el_cookie = ++ec->ec_unique;
+                        if (list_empty(&el->el_chain)) {
+                                list_add(&el->el_chain, &ec->ec_locks);
+                                el->el_cookie = ++ec->ec_unique;
+                        }
+                        atomic_inc(&el->el_refcount);
+                        *cookie = el->el_cookie;

                         spin_unlock(&ec->ec_lock);
                 } else
                         cl_lock_release(env, lck, "ec enqueue", cfs_current());
         }
                         spin_unlock(&ec->ec_lock);
                 } else
                         cl_lock_release(env, lck, "ec enqueue", cfs_current());
         }

+        RETURN(rc);
+}
+
+static int cl_echo_enqueue(struct echo_object *eco, obd_off start, obd_off end,
+                           int mode, __u64 *cookie)
+{
+        struct echo_thread_info *info;
+        struct lu_env *env;
+        struct cl_io *io;
+        int refcheck;
+        int result;
+        ENTRY;
+
+        env = cl_env_get(&refcheck);
+        if (IS_ERR(env))
+                RETURN(PTR_ERR(env));
+
+        info = echo_env_info(env);
+        io = &info->eti_io;
+
+        result = cl_io_init(env, io, CIT_MISC, echo_obj2cl(eco));
+        if (result < 0)
+                GOTO(out, result);
+        LASSERT(result == 0);
+
+        result = cl_echo_enqueue0(env, eco, start, end, mode, cookie, 0);

         cl_io_fini(env, io);
 
         EXIT;
         cl_io_fini(env, io);
 
         EXIT;


@@ -1014,51 +1051,53 @@ out:
         return result;
 }
 
         return result;
 }
 

-static int cl_echo_cancel(struct echo_device *ed, __u64 cookie)
+static int cl_echo_cancel0(struct lu_env *env, struct echo_device *ed,
+                           __u64 cookie)

 {
         struct echo_client_obd *ec = ed->ed_ec;
         struct echo_lock       *ecl = NULL;
         struct list_head       *el;
 {
         struct echo_client_obd *ec = ed->ed_ec;
         struct echo_lock       *ecl = NULL;
         struct list_head       *el;

-        int found = 0;
-        int result;
-
-        struct lu_env *env;
-        int refcheck;
+        int found = 0, still_used = 0;

         ENTRY;
 
         ENTRY;
 

-        env = cl_env_get(&refcheck);
-        if (IS_ERR(env))
-                RETURN(PTR_ERR(env));
-
+        LASSERT(ec != NULL);

         spin_lock (&ec->ec_lock);
         list_for_each (el, &ec->ec_locks) {
                 ecl = list_entry (el, struct echo_lock, el_chain);
                 CDEBUG(D_INFO, "ecl: %p, cookie: %llx\n", ecl, ecl->el_cookie);
                 found = (ecl->el_cookie == cookie);
                 if (found) {
         spin_lock (&ec->ec_lock);
         list_for_each (el, &ec->ec_locks) {
                 ecl = list_entry (el, struct echo_lock, el_chain);
                 CDEBUG(D_INFO, "ecl: %p, cookie: %llx\n", ecl, ecl->el_cookie);
                 found = (ecl->el_cookie == cookie);
                 if (found) {

-                        list_del_init(&ecl->el_chain);
+                        if (atomic_dec_and_test(&ecl->el_refcount))
+                                list_del_init(&ecl->el_chain);
+                        else
+                                still_used = 1;

                         break;
                 }
         }
         spin_unlock (&ec->ec_lock);
 
                         break;
                 }
         }
         spin_unlock (&ec->ec_lock);
 

-        result = -ENOENT;
-        if (found) {
-                struct cl_lock *clk = echo_lock2cl(ecl);
+        if (!found)
+                RETURN(-ENOENT);

 
 

-                cl_lock_get(clk);
-                cl_unuse(env, clk);
-                cl_lock_release(env, clk, "ec enqueue", ecl->el_object);
+        echo_lock_release(env, ecl, still_used);
+        RETURN(0);
+}
+
+static int cl_echo_cancel(struct echo_device *ed, __u64 cookie)
+{
+        struct lu_env *env;
+        int refcheck;
+        int rc;
+        ENTRY;
+
+        env = cl_env_get(&refcheck);
+        if (IS_ERR(env))
+                RETURN(PTR_ERR(env));
+
+        rc = cl_echo_cancel0(env, ed, cookie);

 
 

-                cl_lock_mutex_get(env, clk);
-                cl_lock_cancel(env, clk);
-                cl_lock_delete(env, clk);
-                cl_lock_mutex_put(env, clk);
-                cl_lock_put(env, clk);
-                result = 0;
-        }

         cl_env_put(env, &refcheck);
         cl_env_put(env, &refcheck);

-        RETURN(result);
+        RETURN(rc);

 }
 
 static int cl_echo_async_brw(const struct lu_env *env, struct cl_io *io,
 }
 
 static int cl_echo_async_brw(const struct lu_env *env, struct cl_io *io,


@@ -1089,13 +1128,14 @@ static int cl_echo_object_brw(struct echo_object *eco, int rw, obd_off offset,
         struct cl_2queue        *queue;
         struct cl_io            *io;
         struct cl_page          *clp;
         struct cl_2queue        *queue;
         struct cl_io            *io;
         struct cl_page          *clp;

-
+        struct lustre_handle    lh = { 0 };

         int page_size = cl_page_size(obj);
         int refcheck;
         int rc;
         int i;
         ENTRY;
 
         int page_size = cl_page_size(obj);
         int refcheck;
         int rc;
         int i;
         ENTRY;
 

+        LASSERT((offset & ~CFS_PAGE_MASK) == 0);

         LASSERT(ed->ed_next != NULL);
         env = cl_env_get(&refcheck);
         if (IS_ERR(env))
         LASSERT(ed->ed_next != NULL);
         env = cl_env_get(&refcheck);
         if (IS_ERR(env))


@@ -1111,6 +1151,14 @@ static int cl_echo_object_brw(struct echo_object *eco, int rw, obd_off offset,
                 GOTO(out, rc);
         LASSERT(rc == 0);
 
                 GOTO(out, rc);
         LASSERT(rc == 0);
 

+
+        rc = cl_echo_enqueue0(env, eco, offset,
+                              offset + npages * CFS_PAGE_SIZE - 1,
+                              rw == READ ? LCK_PW : LCK_PW, &lh.cookie,
+                              CILR_NEVER);
+        if (rc < 0)
+                GOTO(error_lock, rc);
+

         for (i = 0; i < npages; i++) {
                 LASSERT(pages[i]);
                 clp = cl_page_find(env, obj, cl_index(obj, offset),
         for (i = 0; i < npages; i++) {
                 LASSERT(pages[i]);
                 clp = cl_page_find(env, obj, cl_index(obj, offset),


@@ -1133,6 +1181,8 @@ static int cl_echo_object_brw(struct echo_object *eco, int rw, obd_off offset,
                 /* drop the reference count for cl_page_find, so that the page
                  * will be freed in cl_2queue_fini. */
                 cl_page_put(env, clp);
                 /* drop the reference count for cl_page_find, so that the page
                  * will be freed in cl_2queue_fini. */
                 cl_page_put(env, clp);

+                cl_page_clip(env, clp, 0, page_size);
+

                 offset += page_size;
         }
 
                 offset += page_size;
         }
 


@@ -1149,12 +1199,13 @@ static int cl_echo_object_brw(struct echo_object *eco, int rw, obd_off offset,
                        async ? "async" : "sync", rc);
         }
 
                        async ? "async" : "sync", rc);
         }
 

+        cl_echo_cancel0(env, ed, lh.cookie);
+        EXIT;
+error_lock:

         cl_2queue_discard(env, io, queue);
         cl_2queue_disown(env, io, queue);
         cl_2queue_fini(env, queue);
         cl_io_fini(env, io);
         cl_2queue_discard(env, io, queue);
         cl_2queue_disown(env, io, queue);
         cl_2queue_fini(env, queue);
         cl_io_fini(env, io);

-
-        EXIT;

 out:
         cl_env_put(env, &refcheck);
         return rc;
 out:
         cl_env_put(env, &refcheck);
         return rc;

diff --git a/lustre/tests/sanity.sh b/lustre/tests/sanity.sh
index e934910..24435a4 100644 (file)
--- a/lustre/tests/sanity.sh
+++ b/lustre/tests/sanity.sh
@@ -6685,6 +6685,71 @@ test_170() {
 }
 run_test 170 "test lctl df to handle corrupted log ====================="
 
 }
 run_test 170 "test lctl df to handle corrupted log ====================="
 

+# it would be good to share it with obdfilter-survey/libecho code
+setup_obdecho_osc () {
+        local rc=0
+        local ost_nid=$1
+        local obdfilter_name=$2
+        [ $rc -eq 0 ] && { $LCTL attach osc ${obdfilter_name}_osc     \
+                           ${obdfilter_name}_osc_UUID || rc=2; }
+        [ $rc -eq 0 ] && { $LCTL --device ${obdfilter_name}_osc setup \
+                           ${obdfilter_name}_UUID  $ost_nid || rc=3; }
+        return $rc
+}
+
+cleaup_obdecho_osc () {
+        local obdfilter_name=$1
+        $LCTL --device ${obdfilter_name}_osc cleanup >/dev/null
+        $LCTL --device ${obdfilter_name}_osc detach  >/dev/null
+        return 0
+}
+
+obdecho_create_test() {
+        local OBD=$1
+        local node=ost
+        local rc=0
+        do_facet $node "$LCTL attach echo_client ec ec_uuid" || rc=1
+        [ $rc -eq 0 ] && { do_facet $node "$LCTL --device ec setup $OBD" ||    \
+                           rc=2; }
+        [ $rc -eq 0 ] && { do_facet $node "$LCTL --device ec create 1" ||      \
+                           rc=3; }
+        [ $rc -eq 0 ] && { do_facet $node "$LCTL --device ec test_brw 0 w 1" ||\
+                           rc=4; }
+        [ $rc -eq 0 -o $rc -gt 2 ] && { do_facet $node "$LCTL --device ec "    \
+                                        "cleanup" || rc=5; }
+        [ $rc -eq 0 -o $rc -gt 1 ] && { do_facet $node "$LCTL --device ec "    \
+                                        "detach" || rc=6; }
+        return $rc
+}
+
+test_180() {
+        local rc=0
+        local rmmod_local=0
+        local rmmod_remote=0
+
+        lsmod | grep -q obdecho || \
+                { load_module obdecho/obdecho && rmmod_local=1; }
+        OBD=$($LCTL dl | grep -v mdt | grep osc | awk '{print $4;exit}')
+        HOST=$($LCTL dl -t | grep -v mdt | grep osc | awk '{print $7;exit}')
+        OBD=`echo $OBD | sed 's/-osc-.*$//'`
+        [ "x$OBD" != "x" ] && { setup_obdecho_osc $HOST $OBD || rc=1; } || rc=1
+        [ $rc -eq 0 ] && { obdecho_create_test ${OBD}_osc || rc=2; }
+        [ "x$OBD" != "x" ] && cleaup_obdecho_osc $OBD
+        [ $rmmod_local -eq 1 ] && rmmod obdecho
+        [ $rc -eq 0 ] || return $rc
+
+        do_facet ost "lsmod | grep -q obdecho || "                      \
+                     "{ insmod ${LUSTRE}/obdecho/obdecho.ko || "        \
+                     "modprobe obdecho; }" && rmmod_remote=1
+        OBD=$(do_facet ost $LCTL dl | awk '/obdfilter/ {print $4;exit}')
+        [ "x$OBD" != "x" ] && { obdecho_create_test $OBD || rc=3; }
+        [ $rmmod_remote -eq 1 ] && do_facet ost "rmmod obdecho"
+        [ $rc -eq 0 ] || return $rc
+
+        true
+}
+run_test 180 "test obdecho ============================================"
+

 # OST pools tests
 POOL=${POOL:-cea1}
 TGT_COUNT=$OSTCOUNT
 # OST pools tests
 POOL=${POOL:-cea1}
 TGT_COUNT=$OSTCOUNT