some cleanup, but theres a slab corruption still

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@2141 29311d96-e01e-0410-9327-a35deaab8ce9

diff --git a/trunk/ceph/kernel/client.c b/trunk/ceph/kernel/client.c
index 93b07fcb4b5d6eae2f06041e6fe29177be6349e2..198dd83edccef31cdf070e8a88502a00f3d45a1d 100644 (file)
--- a/trunk/ceph/kernel/client.c
+++ b/trunk/ceph/kernel/client.c
@@ -249,7 +249,8 @@ void ceph_dispatch(struct ceph_client *client, struct ceph_msg *msg)
 
                /* mds client */
        case CEPH_MSG_MDS_MAP:
-               ceph_mdsc_handle_map(&client->mdsc, msg);
+               //ceph_mdsc_handle_map(&client->mdsc, msg);
+               ceph_msg_put(msg);
                break;
        case CEPH_MSG_CLIENT_REPLY:
                ceph_mdsc_handle_reply(&client->mdsc, msg);

diff --git a/trunk/ceph/kernel/ktcp.c b/trunk/ceph/kernel/ktcp.c
index c5c9189eb62fc88c661a6ac1496cf63875281efb..bbd3ddae2dfb43beac9643ee13ab25dbe8124aad 100644 (file)
--- a/trunk/ceph/kernel/ktcp.c
+++ b/trunk/ceph/kernel/ktcp.c
@@ -222,11 +222,11 @@ int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
        struct msghdr msg = {.msg_flags = 0};
        int rlen = 0;           /* length read */
 
-       dout(30, "ceph_tcp_recvmsg %p len %d\n", sock, (int)len);
+       //dout(30, "ceph_tcp_recvmsg %p len %d\n", sock, (int)len);
        msg.msg_flags |= MSG_DONTWAIT | MSG_NOSIGNAL;
        /* receive one kvec for now...  */
        rlen = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
-       dout(30, "ceph_tcp_recvmsg %p len %d ret = %d\n", sock, (int)len, rlen);
+       //dout(30, "ceph_tcp_recvmsg %p len %d ret = %d\n", sock, (int)len, rlen);
        return(rlen);
 }
 
@@ -238,10 +238,10 @@ int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov, size_t kvlen, size_t
        struct msghdr msg = {.msg_flags = 0};
        int rlen = 0;
 
-       dout(30, "ceph_tcp_sendmsg %p len %d\n", sock, (int)len);
+       //dout(30, "ceph_tcp_sendmsg %p len %d\n", sock, (int)len);
        msg.msg_flags |=  MSG_DONTWAIT | MSG_NOSIGNAL;
        rlen = kernel_sendmsg(sock, &msg, iov, kvlen, len);
-       dout(30, "ceph_tcp_sendmsg %p len %d ret = %d\n", sock, (int)len, rlen);
+       //dout(30, "ceph_tcp_sendmsg %p len %d ret = %d\n", sock, (int)len, rlen);
        return(rlen);
 }
 

diff --git a/trunk/ceph/kernel/mds_client.c b/trunk/ceph/kernel/mds_client.c
index 7cb981d62dbb28e468aee5593109cde557f54cff..807783eb7b76009ad17dda9d31fb1aa089c66091 100644 (file)
--- a/trunk/ceph/kernel/mds_client.c
+++ b/trunk/ceph/kernel/mds_client.c
@@ -83,15 +83,27 @@ static int choose_mds(struct ceph_mds_client *mdsc, struct ceph_mds_request *req
 
 static void register_session(struct ceph_mds_client *mdsc, int mds)
 {
+       struct ceph_mds_session *s;
+
        /* register */
        if (mds >= mdsc->max_sessions) {
                /* realloc */
+               struct ceph_mds_session **sa;
+               sa = kzalloc(mds * sizeof(struct ceph_mds_session), GFP_KERNEL);
+               BUG_ON(sa == NULL);  /* i am lazy */
+               if (mdsc->sessions) {
+                       memcpy(sa, mdsc->sessions, 
+                              mdsc->max_sessions*sizeof(struct ceph_mds_session));
+                       kfree(mdsc->sessions);
+               }
+               mdsc->sessions = sa;
        }
-       mdsc->sessions[mds] = kmalloc(sizeof(struct ceph_mds_session), GFP_KERNEL);
-       mdsc->sessions[mds]->s_state = 0;
-       mdsc->sessions[mds]->s_cap_seq = 0;
-       init_completion(&mdsc->sessions[mds]->s_completion);
-       atomic_set(&mdsc->sessions[mds]->s_ref, 1);
+       s = kmalloc(sizeof(struct ceph_mds_session), GFP_KERNEL);
+       s->s_state = 0;
+       s->s_cap_seq = 0;
+       init_completion(&s->s_completion);
+       atomic_set(&s->s_ref, 1);
+       mdsc->sessions[mds] = s;
 }
 
 static struct ceph_mds_session *get_session(struct ceph_mds_client *mdsc, int mds)
@@ -223,7 +235,7 @@ void ceph_mdsc_init(struct ceph_mds_client *mdsc, struct ceph_client *client)
 {
        spin_lock_init(&mdsc->lock);
        mdsc->client = client;
-       mdsc->mdsmap = 0;  /* none yet */
+       mdsc->mdsmap = 0;            /* none yet */
        mdsc->sessions = 0;
        mdsc->max_sessions = 0;
        mdsc->last_tid = 0;
@@ -640,7 +652,8 @@ void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
        /* do we need it? */
        spin_lock(&mdsc->lock);
        if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
-               dout(2, "ceph_mdsc_handle_map epoch %llu < our %llu\n", epoch, mdsc->mdsmap->m_epoch);
+               dout(2, "ceph_mdsc_handle_map epoch %llu < our %llu\n", 
+                    epoch, mdsc->mdsmap->m_epoch);
                spin_unlock(&mdsc->lock);
                goto out;
        }
@@ -658,7 +671,8 @@ void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
                        oldmap = mdsc->mdsmap;
                        mdsc->mdsmap = newmap;
                        spin_unlock(&mdsc->lock);
-                       ceph_mdsmap_destroy(oldmap);
+                       if (oldmap)
+                               ceph_mdsmap_destroy(oldmap);
                } else {
                        spin_unlock(&mdsc->lock);
                        dout(2, "ceph_mdsc_handle_map lost decode race?\n");

diff --git a/trunk/ceph/kernel/mdsmap.c b/trunk/ceph/kernel/mdsmap.c
index 4e4215484bb59e027d124480796aa194786cef5f..35ecc7a5bb40c4a56e72d6a38c16e868cade2f87 100644 (file)
--- a/trunk/ceph/kernel/mdsmap.c
+++ b/trunk/ceph/kernel/mdsmap.c
@@ -81,6 +81,8 @@ struct ceph_mdsmap *ceph_mdsmap_decode(void **p, void *end)
        for (i=0; i<n; i++) {
                if ((err = ceph_decode_32(p, end, &mds)) != 0)
                        goto bad;
+               if (mds >= m->m_max_mds)
+                       goto bad;
                if ((err = ceph_decode_32(p, end, &m->m_state[mds])) != 0)
                        goto bad;
        }
@@ -96,12 +98,15 @@ struct ceph_mdsmap *ceph_mdsmap_decode(void **p, void *end)
        for (i=0; i<n; i++) {
                if ((err = ceph_decode_32(p, end, &mds)) != 0)
                        goto bad;
+               if (mds >= m->m_max_mds)
+                       goto bad;
                *p += sizeof(struct ceph_entity_name);
                if ((err = ceph_decode_addr(p, end, &m->m_addr[mds])) != 0)
                        goto bad;
        }
 
        /* ok, we don't care about the rest. */
+       dout(30, "mdsmap_decode success epoch %llu\n", m->m_epoch);
        return m;
 
 bad:

diff --git a/trunk/ceph/kernel/messenger.c b/trunk/ceph/kernel/messenger.c
index d7cfe5f3fde57538c5bea9f0c07b9cd1543d3d13..2681576dccf402c6e7d0a6573b702480ed6cc73e 100644 (file)
--- a/trunk/ceph/kernel/messenger.c
+++ b/trunk/ceph/kernel/messenger.c
@@ -234,7 +234,8 @@ static int write_partial_kvec(struct ceph_connection *con)
        dout(30, "write_partial_kvec %p left %d vec %d bytes\n", con, 
             con->out_kvec_left, con->out_kvec_bytes);
        while (con->out_kvec_bytes > 0) {
-               ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur, con->out_kvec_left, con->out_kvec_bytes);
+               ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur, 
+                                      con->out_kvec_left, con->out_kvec_bytes);
                if (ret <= 0) goto out;
                con->out_kvec_bytes -= ret;
                if (con->out_kvec_bytes == 0)
@@ -295,7 +296,7 @@ static void prepare_write_message(struct ceph_connection *con)
        /* move to sending/sent list */
        list_del(&m->list_head);
        list_add_tail(&m->list_head, &con->out_sent);
-       con->out_msg = m;
+       con->out_msg = m;  /* FIXME: do we want to take a reference here? */
 
        /* encode header */
        ceph_encode_header(&con->out_hdr, &m->hdr);
@@ -443,6 +444,7 @@ done:
  */
 static int prepare_read_message(struct ceph_connection *con)
 {
+       BUG_ON(con->in_msg != NULL);
        con->in_tag = CEPH_MSGR_TAG_MSG;
        con->in_base_pos = 0;
        con->in_msg = kzalloc(sizeof(*con->in_msg), GFP_KERNEL);
@@ -485,6 +487,7 @@ static int read_message_partial(struct ceph_connection *con)
                        m->front.iov_base = kmalloc(m->hdr.front_len, GFP_KERNEL);
                        if (m->front.iov_base == NULL)
                                return -ENOMEM;
+                       dout(50, "front is %p\n", m->front.iov_base);
                }
                left = m->hdr.front_len - m->front.iov_len;
                ret = ceph_tcp_recvmsg(con->sock, (char*)m->front.iov_base + m->front.iov_len, left);
@@ -633,20 +636,24 @@ static void process_connect(struct ceph_connection *con)
 static int read_accept_partial(struct ceph_connection *con)
 {
        int ret;
+       int to;
 
        /* peer addr */
-       while (con->in_base_pos < sizeof(con->peer_addr)) {
-               int left = sizeof(con->peer_addr) - con->in_base_pos;
-               ret = ceph_tcp_recvmsg(con->sock, (char*)&con->peer_addr + con->in_base_pos, left);
+       to = sizeof(con->peer_addr);
+       while (con->in_base_pos < to) {
+               int left = to - con->in_base_pos;
+               int have = con->in_base_pos;
+               ret = ceph_tcp_recvmsg(con->sock, (char*)&con->peer_addr + have, left);
                if (ret <= 0) return ret;
                con->in_base_pos += ret;
        }
 
        /* connect_seq */
-       while (con->in_base_pos < sizeof(con->peer_addr) + sizeof(con->connect_seq)) {
-               int off = con->in_base_pos - sizeof(con->peer_addr);
-               int left = sizeof(con->peer_addr) + sizeof(con->connect_seq) - con->in_base_pos;
-               ret = ceph_tcp_recvmsg(con->sock, (char*)&con->connect_seq + off, left);
+       to += sizeof(con->connect_seq);
+       while (con->in_base_pos < to) {
+               int left = to - con->in_base_pos;
+               int have = sizeof(con->peer_addr) - left;
+               ret = ceph_tcp_recvmsg(con->sock, (char*)&con->connect_seq + have, left);
                if (ret <= 0) return ret;
                con->in_base_pos += ret;
        }
@@ -665,6 +672,7 @@ static void process_accept(struct ceph_connection *con)
        existing = get_connection(con->msgr, &con->peer_addr);
        if (existing) {
                spin_lock(&existing->lock);
+               /* replace existing connection? */
                if ((test_bit(CONNECTING, &existing->state) && 
                     compare_addr(&con->msgr->inst.addr, &con->peer_addr)) ||
                    (test_bit(OPEN, &existing->state) && 
@@ -834,9 +842,9 @@ struct ceph_messenger *ceph_messenger_create(struct ceph_entity_addr *myaddr)
 
        /* create listening socket */
        ret = ceph_tcp_listen(msgr);
-       if(ret < 0) {
+       if (ret < 0) {
                kfree(msgr);
-               return  ERR_PTR(ret);
+               return ERR_PTR(ret);
        }
        if (myaddr) 
                msgr->inst.addr.ipaddr.sin_addr = myaddr->ipaddr.sin_addr;
@@ -949,6 +957,7 @@ struct ceph_msg *ceph_msg_new(int type, int front_len, int page_len, int page_of
        m->front.iov_base = kmalloc(front_len, GFP_KERNEL);
        if (m->front.iov_base == NULL)
                goto out2;
+       dout(50, "ceph_msg_new front is %p\n", m->front.iov_base);
        m->front.iov_len = front_len;
 
        /* pages */
@@ -981,11 +990,12 @@ void ceph_msg_put(struct ceph_msg *m)
                if (m->pages) {
                        for (i=0; i<m->nr_pages; i++)
                                if (m->pages[i])
-                                       kfree(m->pages[i]);
+                                       __free_pages(m->pages[i], 0);
                        kfree(m->pages);
                }
-               if (m->front.iov_base)
+               if (m->front.iov_base) {
                        kfree(m->front.iov_base);
+               }
                kfree(m);
        }
 }

diff --git a/trunk/ceph/kernel/osd_client.c b/trunk/ceph/kernel/osd_client.c
index b3d8d22d29c5067caefc15eb7b190212722d1c5e..f012e655ec8b55bd807f89ce0eb7b3fdc017c40f 100644 (file)
--- a/trunk/ceph/kernel/osd_client.c
+++ b/trunk/ceph/kernel/osd_client.c
@@ -149,6 +149,7 @@ static struct ceph_osdmap *apply_incremental(void **p, void *end, struct ceph_os
                goto bad;
        if ((err = ceph_decode_64(p, end, &epoch)) < 0)
                goto bad;
+       BUG_ON(epoch != map->epoch+1);
        if ((err = ceph_decode_64(p, end, &mon_epoch)) < 0)
                goto bad;
        if ((err = ceph_decode_32(p, end, &ctime.tv_sec)) < 0)
@@ -163,8 +164,6 @@ static struct ceph_osdmap *apply_incremental(void **p, void *end, struct ceph_os
                newmap = osdmap_decode(p, min(*p+len, end));
                return newmap;  /* error or not */
        }
-       if (!map || epoch != map->epoch+1) 
-               return 0; /* old or new, or no existing; done */
 
        /* new crush? */
        if ((err = ceph_decode_32(p, end, &len)) < 0)
@@ -246,6 +245,7 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 {
        void *p, *end, *next;
        __u32 nr_maps, maplen;
+       __u64 epoch;
        struct ceph_osdmap *newmap = 0;
        int err;
 
@@ -257,19 +257,25 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
        if ((err = ceph_decode_32(&p, end, &nr_maps)) < 0)
                goto bad;
        while (nr_maps--) {
+               if ((err = ceph_decode_64(&p, end, &epoch)) < 0)
+                       goto bad;
                if ((err = ceph_decode_32(&p, end, &maplen)) < 0)
                        goto bad;
                next = p + maplen;
-               newmap = apply_incremental(p, min(p+maplen,end), osdc->osdmap);
-               if (IS_ERR(newmap)) {
-                       err = PTR_ERR(newmap);
-                       goto bad;
-               }
-               if (newmap != osdc->osdmap) {
-                       osdmap_destroy(osdc->osdmap);
-                       osdc->osdmap = newmap;
+               if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
+                       newmap = apply_incremental(p, min(p+maplen,end), osdc->osdmap);
+                       if (IS_ERR(newmap)) {
+                               err = PTR_ERR(newmap);
+                               goto bad;
+                       }
+                       if (newmap != osdc->osdmap) {
+                               osdmap_destroy(osdc->osdmap);
+                               osdc->osdmap = newmap;
+                       }
+                       dout(1, "applied incremental map %llu\n", epoch);
+               } else {
+                       dout(1, "ignored incremental map %llu\n", epoch);
                }
-               dout(1, "got incremental map %llu\n", newmap->epoch);
                p = next;
        }
        if (newmap) 
@@ -279,22 +285,32 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
        if ((err = ceph_decode_32(&p, end, &nr_maps)) < 0)
                goto bad;
        while (nr_maps > 1) {
-               dout(5, "ignoring non-latest full map\n");
+               dout(5, "skipping non-latest full map\n");
+               if ((err = ceph_decode_64(&p, end, &epoch)) < 0)
+                       goto bad;
                if ((err = ceph_decode_32(&p, end, &maplen)) < 0)
                        goto bad;
                p += maplen;
        }
        if (nr_maps) {
-               if ((err = ceph_decode_32(&p, end, &maplen)) < 0)
+               if ((err = ceph_decode_64(&p, end, &epoch)) < 0)
                        goto bad;
-               newmap = osdmap_decode(&p, min(p+maplen,end));
-               if (IS_ERR(newmap)) {
-                       err = PTR_ERR(newmap);
+               if ((err = ceph_decode_32(&p, end, &maplen)) < 0)
                        goto bad;
+               if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
+                       dout(10, "full map %llu is older than our %llu\n", 
+                            epoch, osdc->osdmap->epoch);
+               } else {
+                       newmap = osdmap_decode(&p, min(p+maplen,end));
+                       if (IS_ERR(newmap)) {
+                               err = PTR_ERR(newmap);
+                               goto bad;
+                       }
+                       if (osdc->osdmap)
+                               osdmap_destroy(osdc->osdmap);
+                       osdc->osdmap = newmap;
+                       dout(10, "took full map %llu\n", newmap->epoch);
                }
-               osdmap_destroy(osdc->osdmap);
-               osdc->osdmap = newmap;
-               dout(1, "got full map %llu\n", newmap->epoch);
        }
        
 out:

diff --git a/trunk/ceph/kernel/super.c b/trunk/ceph/kernel/super.c
index ccfa7a3d11e6eb562152d09a6d7bd9656de05e36..627ae3b9feaa4ffe49c06b0cb5d27b8cd4da0090 100644 (file)
--- a/trunk/ceph/kernel/super.c
+++ b/trunk/ceph/kernel/super.c
@@ -145,11 +145,11 @@ static int ceph_set_super(struct super_block *s, void *data)
        
        ret = set_anon_super(s, 0);  /* what is the second arg for? */
        if (ret != 0)
-               goto out; 
+               goto bail; 
        
        return ret;
 
-out:
+bail:
        kfree(s->s_fs_info);
        s->s_fs_info = 0;
        return ret;