| 914 | } |
| 915 | |
| 916 | Status Statestore::OfferUpdate(const ScheduledSubscriberUpdate& update, |
| 917 | ThreadPool<ScheduledSubscriberUpdate>* threadpool) { |
| 918 | // Somewhat confusingly, we're checking the number of entries in a particular |
| 919 | // threadpool's work queue to decide whether or not we have too many |
| 920 | // subscribers. The number of subscribers registered can be actually more |
| 921 | // than statestore_max_subscribers. This is because RegisterSubscriber() adds |
| 922 | // the new subscriber to subscribers_ first before scheduling its updates. |
| 923 | // Should we be stricter in enforcing this limit on subscribers_.size() itself? |
| 924 | if (threadpool->GetQueueSize() >= FLAGS_statestore_max_subscribers |
| 925 | || !threadpool->Offer(update)) { |
| 926 | stringstream ss; |
| 927 | ss << "Maximum subscriber limit reached: " << FLAGS_statestore_max_subscribers; |
| 928 | ss << ", subscribers_ size: " << subscribers_.size(); |
| 929 | SubscriberMap::iterator subscriber_it = subscribers_.find(update.subscriber_id); |
| 930 | DCHECK(subscriber_it != subscribers_.end()); |
| 931 | subscribers_.erase(subscriber_it); |
| 932 | if (FLAGS_enable_statestored_ha) { |
| 933 | ActiveConnStateMap::iterator conn_states_it = |
| 934 | active_conn_states_.find(update.subscriber_id); |
| 935 | if (conn_states_it != active_conn_states_.end()) { |
| 936 | if (conn_states_it->second == TStatestoreConnState::FAILED) { |
| 937 | --failed_conn_state_count_; |
| 938 | } |
| 939 | active_conn_states_.erase(conn_states_it); |
| 940 | } |
| 941 | } |
| 942 | LOG(ERROR) << ss.str(); |
| 943 | return Status(ss.str()); |
| 944 | } |
| 945 | |
| 946 | return Status::OK(); |
| 947 | } |
| 948 | |
| 949 | Status Statestore::RegisterSubscriber(const SubscriberId& subscriber_id, |
| 950 | const TNetworkAddress& location, |