store->time_log_prepare_entry(entry, ut, section, bucket.name, bl);
+ m[index].first.push_back(bucket.name);
m[index].second.push_back(entry);
}
ChangeStatusPtr status;
_get_change(bucket_name, status);
+ ldout(cct, 20) << "RGWDataChangesLog::update_renewd() bucket_name=" << bucket_name << " expiration=" << expiration << dendl;
status->cur_expiration = expiration;
}
status->lock->Lock();
+ ldout(cct, 20) << "RGWDataChangesLog::add_entry() bucket.name=" << bucket.name << " now=" << now << " cur_expiration=" << status->cur_expiration << dendl;
+
if (now < status->cur_expiration) {
/* no need to send, recently completed */
status->lock->Unlock();
status->cond = new RefCountedCond;
status->pending = true;
- status->cur_sent = now;
+ string& oid = oids[choose_oid(bucket)];
+ utime_t expiration;
- status->lock->Unlock();
+ int ret;
+
+ do {
+ status->cur_sent = now;
+
+ expiration = now;
+ expiration += utime_t(cct->_conf->rgw_data_log_window, 0);
+
+ status->lock->Unlock();
- string& oid = oids[choose_oid(bucket)];
+ bufferlist bl;
+ rgw_data_change change;
+ change.entity_type = ENTITY_TYPE_BUCKET;
+ change.key = bucket.name;
+ ::encode(change, bl);
+ string section;
- utime_t ut = ceph_clock_now(cct);
- bufferlist bl;
- rgw_data_change change;
- change.entity_type = ENTITY_TYPE_BUCKET;
- change.key = bucket.name;
- ::encode(change, bl);
- string section;
- int ret = store->time_log_add(oid, ut, section, change.key, bl);
+ ldout(cct, 20) << "RGWDataChangesLog::add_entry() sending update with now=" << now << " cur_expiration=" << expiration << dendl;
- status->lock->Lock();
+ ret = store->time_log_add(oid, now, section, change.key, bl);
+
+ now = ceph_clock_now(cct);
+
+ status->lock->Lock();
+
+ } while (!ret && ceph_clock_now(cct) > expiration);
cond = status->cond;