X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=ofproto%2Fofproto-dpif-monitor.c;h=764c7b19939595dc8769cf005814c45302ec0be2;hb=HEAD;hp=33115f3344e68ccd5218e8acdb863173e5d9ca25;hpb=4b0424809b823101c969a0691fc1db0c880ae64a;p=sliver-openvswitch.git diff --git a/ofproto/ofproto-dpif-monitor.c b/ofproto/ofproto-dpif-monitor.c index 33115f334..764c7b199 100644 --- a/ofproto/ofproto-dpif-monitor.c +++ b/ofproto/ofproto-dpif-monitor.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009, 2010, 2011, 2012, 2013 Nicira, Inc. + * Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ #include "bfd.h" #include "cfm.h" +#include "guarded-list.h" #include "hash.h" #include "heap.h" #include "hmap.h" @@ -52,37 +53,51 @@ struct mport { uint8_t hw_addr[OFP_ETH_ALEN]; /* Hardware address. */ }; +/* Entry of the 'send_soon' list. Contains the pointer to the + * 'ofport_dpif'. Note, the pointed object is not protected, so + * users should always use the mport_find() to convert it to 'mport'. */ +struct send_soon_entry { + struct list list_node; /* In send_soon. */ + const struct ofport_dpif *ofport; +}; + /* hmap that contains "struct mport"s. */ static struct hmap monitor_hmap = HMAP_INITIALIZER(&monitor_hmap); /* heap for ordering mport based on bfd/cfm wakeup time. */ static struct heap monitor_heap; +/* guarded-list for storing the mports that need to send bfd/cfm control + * packet soon. */ +static struct guarded_list send_soon = GUARDED_LIST_INITIALIZER(&send_soon); + /* The monitor thread id. */ static pthread_t monitor_tid; /* True if the monitor thread is running. */ static bool monitor_running; static struct latch monitor_exit_latch; -static struct ovs_rwlock monitor_rwlock = OVS_RWLOCK_INITIALIZER; +static struct ovs_mutex monitor_mutex = OVS_MUTEX_INITIALIZER; static void *monitor_main(void *); +static void monitor_check_send_soon(struct ofpbuf *); static void monitor_run(void); +static void monitor_mport_run(struct mport *, struct ofpbuf *); static void mport_register(const struct ofport_dpif *, struct bfd *, struct cfm *, uint8_t[ETH_ADDR_LEN]) - OVS_REQ_WRLOCK(monitor_rwlock); + OVS_REQUIRES(monitor_mutex); static void mport_unregister(const struct ofport_dpif *) - OVS_REQ_WRLOCK(monitor_rwlock); + OVS_REQUIRES(monitor_mutex); static void mport_update(struct mport *, struct bfd *, struct cfm *, - uint8_t[ETH_ADDR_LEN]) OVS_REQ_WRLOCK(monitor_rwlock); + uint8_t[ETH_ADDR_LEN]) OVS_REQUIRES(monitor_mutex); static struct mport *mport_find(const struct ofport_dpif *) - OVS_REQ_WRLOCK(monitor_rwlock); + OVS_REQUIRES(monitor_mutex); /* Tries finding and returning the 'mport' from the monitor_hmap. * If there is no such 'mport', returns NULL. */ static struct mport * -mport_find(const struct ofport_dpif *ofport) OVS_REQ_WRLOCK(monitor_rwlock) +mport_find(const struct ofport_dpif *ofport) OVS_REQUIRES(monitor_mutex) { struct mport *node; @@ -100,7 +115,7 @@ mport_find(const struct ofport_dpif *ofport) OVS_REQ_WRLOCK(monitor_rwlock) static void mport_register(const struct ofport_dpif *ofport, struct bfd *bfd, struct cfm *cfm, uint8_t *hw_addr) - OVS_REQ_WRLOCK(monitor_rwlock) + OVS_REQUIRES(monitor_mutex) { struct mport *mport = mport_find(ofport); @@ -116,7 +131,7 @@ mport_register(const struct ofport_dpif *ofport, struct bfd *bfd, /* Removes mport from monitor_hmap and monitor_heap and frees it. */ static void mport_unregister(const struct ofport_dpif *ofport) - OVS_REQ_WRLOCK(monitor_rwlock) + OVS_REQUIRES(monitor_mutex) { struct mport *mport = mport_find(ofport); @@ -131,7 +146,7 @@ mport_unregister(const struct ofport_dpif *ofport) /* Updates the fields of an existing mport struct. */ static void mport_update(struct mport *mport, struct bfd *bfd, struct cfm *cfm, - uint8_t hw_addr[ETH_ADDR_LEN]) OVS_REQ_WRLOCK(monitor_rwlock) + uint8_t hw_addr[ETH_ADDR_LEN]) OVS_REQUIRES(monitor_mutex) { ovs_assert(mport); @@ -158,7 +173,6 @@ mport_update(struct mport *mport, struct bfd *bfd, struct cfm *cfm, static void * monitor_main(void * args OVS_UNUSED) { - set_subprogram_name("monitor"); VLOG_INFO("monitor thread created"); while (!latch_is_set(&monitor_exit_latch)) { monitor_run(); @@ -173,9 +187,8 @@ monitor_main(void * args OVS_UNUSED) * reconfigured monitoring ports are run in a timely manner. */ #define MONITOR_INTERVAL_MSEC 100 -/* Checks the sending of control packets on mports that have timed out. - * Sends the control packets if needed. Executes bfd and cfm periodic - * functions (run, wait) on those mports. */ +/* Checks the 'send_soon' list and the heap for mports that have timed + * out bfd/cfm sessions. */ static void monitor_run(void) { @@ -184,40 +197,27 @@ monitor_run(void) struct ofpbuf packet; ofpbuf_use_stub(&packet, stub, sizeof stub); - ovs_rwlock_wrlock(&monitor_rwlock); + ovs_mutex_lock(&monitor_mutex); + + /* The monitor_check_send_soon() needs to be run twice. The first + * time is for preventing the same 'mport' from being processed twice + * (i.e. once from heap, the other from the 'send_soon' array). + * The second run is to cover the case when the control packet is sent + * via patch port and the other end needs to send back immediately. */ + monitor_check_send_soon(&packet); + prio_now = MSEC_TO_PRIO(time_msec()); /* Peeks the top of heap and checks if we should run this mport. */ while (!heap_is_empty(&monitor_heap) && heap_max(&monitor_heap)->priority >= prio_now) { - long long int next_wake_time; struct mport *mport; mport = CONTAINER_OF(heap_max(&monitor_heap), struct mport, heap_node); - if (mport->cfm && cfm_should_send_ccm(mport->cfm)) { - ofpbuf_clear(&packet); - cfm_compose_ccm(mport->cfm, &packet, mport->hw_addr); - ofproto_dpif_send_packet(mport->ofport, &packet); - } - if (mport->bfd && bfd_should_send_packet(mport->bfd)) { - ofpbuf_clear(&packet); - bfd_put_packet(mport->bfd, &packet, mport->hw_addr); - ofproto_dpif_send_packet(mport->ofport, &packet); - } - if (mport->cfm) { - cfm_run(mport->cfm); - cfm_wait(mport->cfm); - } - if (mport->bfd) { - bfd_run(mport->bfd); - bfd_wait(mport->bfd); - } - /* Computes the next wakeup time for this mport. */ - next_wake_time = MIN(bfd_wake_time(mport->bfd), - cfm_wake_time(mport->cfm)); - heap_change(&monitor_heap, &mport->heap_node, - MSEC_TO_PRIO(next_wake_time)); + monitor_mport_run(mport, &packet); } + monitor_check_send_soon(&packet); + /* Waits on the earliest next wakeup time. */ if (!heap_is_empty(&monitor_heap)) { long long int next_timeout, next_mport_wakeup; @@ -226,9 +226,64 @@ monitor_run(void) next_mport_wakeup = PRIO_TO_MSEC(heap_max(&monitor_heap)->priority); poll_timer_wait_until(MIN(next_timeout, next_mport_wakeup)); } - ovs_rwlock_unlock(&monitor_rwlock); + ovs_mutex_unlock(&monitor_mutex); ofpbuf_uninit(&packet); } + +/* Checks the 'send_soon' list for any mport that needs to send cfm/bfd + * control packet immediately, and calls monitor_mport_run(). */ +static void +monitor_check_send_soon(struct ofpbuf *packet) + OVS_REQUIRES(monitor_mutex) +{ + while (!guarded_list_is_empty(&send_soon)) { + struct send_soon_entry *entry; + struct mport *mport; + + entry = CONTAINER_OF(guarded_list_pop_front(&send_soon), + struct send_soon_entry, list_node); + mport = mport_find(entry->ofport); + if (mport) { + monitor_mport_run(mport, packet); + } + free(entry); + } +} + +/* Checks the sending of control packet on 'mport'. Sends the control + * packet if needed. Executes bfd and cfm periodic functions (run, wait) + * on 'mport'. And changes the location of 'mport' in heap based on next + * timeout. */ +static void +monitor_mport_run(struct mport *mport, struct ofpbuf *packet) + OVS_REQUIRES(monitor_mutex) +{ + long long int next_wake_time; + + if (mport->cfm && cfm_should_send_ccm(mport->cfm)) { + ofpbuf_clear(packet); + cfm_compose_ccm(mport->cfm, packet, mport->hw_addr); + ofproto_dpif_send_packet(mport->ofport, packet); + } + if (mport->bfd && bfd_should_send_packet(mport->bfd)) { + ofpbuf_clear(packet); + bfd_put_packet(mport->bfd, packet, mport->hw_addr); + ofproto_dpif_send_packet(mport->ofport, packet); + } + if (mport->cfm) { + cfm_run(mport->cfm); + cfm_wait(mport->cfm); + } + if (mport->bfd) { + bfd_run(mport->bfd); + bfd_wait(mport->bfd); + } + /* Computes the next wakeup time for this mport. */ + next_wake_time = MIN(bfd_wake_time(mport->bfd), + cfm_wake_time(mport->cfm)); + heap_change(&monitor_heap, &mport->heap_node, + MSEC_TO_PRIO(next_wake_time)); +} /* Creates the mport in monitor module if either bfd or cfm @@ -240,20 +295,20 @@ ofproto_dpif_monitor_port_update(const struct ofport_dpif *ofport, struct bfd *bfd, struct cfm *cfm, uint8_t hw_addr[ETH_ADDR_LEN]) { - ovs_rwlock_wrlock(&monitor_rwlock); + ovs_mutex_lock(&monitor_mutex); if (!cfm && !bfd) { mport_unregister(ofport); } else { mport_register(ofport, bfd, cfm, hw_addr); } - ovs_rwlock_unlock(&monitor_rwlock); + ovs_mutex_unlock(&monitor_mutex); /* If the monitor thread is not running and the hmap * is not empty, starts it. If it is and the hmap is empty, * terminates it. */ if (!monitor_running && !hmap_is_empty(&monitor_hmap)) { latch_init(&monitor_exit_latch); - xpthread_create(&monitor_tid, NULL, monitor_main, NULL); + monitor_tid = ovs_thread_create("monitor", monitor_main, NULL); monitor_running = true; } else if (monitor_running && hmap_is_empty(&monitor_hmap)) { latch_set(&monitor_exit_latch); @@ -263,25 +318,16 @@ ofproto_dpif_monitor_port_update(const struct ofport_dpif *ofport, } } -/* Moves the mport on top of the heap. This is necessary when - * for example, bfd POLL is received and the mport should - * immediately send FINAL back. */ -void -ofproto_dpif_monitor_port_send_soon_safe(const struct ofport_dpif *ofport) -{ - ovs_rwlock_wrlock(&monitor_rwlock); - ofproto_dpif_monitor_port_send_soon(ofport); - ovs_rwlock_unlock(&monitor_rwlock); -} - +/* Registers the 'ofport' in the 'send_soon' list. We cannot directly + * insert the corresponding mport to the 'send_soon' list, since the + * 'send_soon' list is not updated when the mport is removed. + * + * Reader of the 'send_soon' list is responsible for freeing the entry. */ void ofproto_dpif_monitor_port_send_soon(const struct ofport_dpif *ofport) - OVS_REQ_WRLOCK(monitor_rwlock) { - struct mport *mport; + struct send_soon_entry *entry = xzalloc(sizeof *entry); + entry->ofport = ofport; - mport = mport_find(ofport); - if (mport) { - heap_change(&monitor_heap, &mport->heap_node, LLONG_MAX); - } + guarded_list_push_back(&send_soon, &entry->list_node, SIZE_MAX); }