entry->ue_key = key;
atomic_set(&entry->ue_refcount, 0);
init_waitqueue_head(&entry->ue_waitq);
+ entry->ue_acquire_expire = 0;
+ entry->ue_expire = 0;
if (cache->uc_ops->init_entry)
cache->uc_ops->init_entry(entry, args);
return entry;
}
}
+static inline void write_lock_from_read(rwlock_t *lock, bool *writelock)
+{
+ if (!*writelock) {
+ read_unlock(lock);
+ write_lock(lock);
+ *writelock = true;
+ }
+}
+
static int check_unlink_entry(struct upcall_cache *cache,
- struct upcall_cache_entry *entry)
+ struct upcall_cache_entry *entry,
+ bool writelock)
{
time64_t now = ktime_get_seconds();
now < entry->ue_acquire_expire)
return 0;
- UC_CACHE_SET_EXPIRED(entry);
- wake_up(&entry->ue_waitq);
- } else if (!UC_CACHE_IS_INVALID(entry)) {
+ if (writelock) {
+ UC_CACHE_SET_EXPIRED(entry);
+ wake_up(&entry->ue_waitq);
+ }
+ } else if (!UC_CACHE_IS_INVALID(entry) && writelock) {
UC_CACHE_SET_EXPIRED(entry);
}
- list_del_init(&entry->ue_hash);
- if (!atomic_read(&entry->ue_refcount))
- free_entry(cache, entry);
+ if (writelock) {
+ list_del_init(&entry->ue_hash);
+ if (!atomic_read(&entry->ue_refcount))
+ free_entry(cache, entry);
+ }
return 1;
}
+int upcall_cache_set_upcall(struct upcall_cache *cache, const char *buffer,
+ size_t count, bool path_only)
+{
+ char *upcall;
+
+ if (count >= UC_CACHE_UPCALL_MAXPATH)
+ return -E2BIG;
+
+ OBD_ALLOC(upcall, count + 1);
+ if (upcall == NULL)
+ return -ENOMEM;
+
+ /* Remove any extraneous bits from the upcall (e.g. linefeeds) */
+ if (sscanf(buffer, "%s", upcall) != 1)
+ goto invalid;
+
+ if (upcall[0] == '/')
+ goto valid;
+
+ if (path_only)
+ goto invalid;
+
+ if (strcasecmp(upcall, "NONE") == 0) {
+ snprintf(upcall, count + 1, "NONE");
+ goto valid;
+ }
+
+invalid:
+ OBD_FREE(upcall, count + 1);
+ return -EINVAL;
+
+valid:
+ down_write(&cache->uc_upcall_rwsem);
+ strcpy(cache->uc_upcall, upcall);
+ up_write(&cache->uc_upcall_rwsem);
+
+ OBD_FREE(upcall, count + 1);
+ return 0;
+}
+EXPORT_SYMBOL(upcall_cache_set_upcall);
+
static inline int refresh_entry(struct upcall_cache *cache,
struct upcall_cache_entry *entry)
{
bool failedacquiring = false;
struct list_head *head;
wait_queue_entry_t wait;
+ bool writelock;
int rc, found;
+
ENTRY;
LASSERT(cache);
cache->uc_hashsize)];
find_again:
found = 0;
- spin_lock(&cache->uc_lock);
+ if (new) {
+ write_lock(&cache->uc_lock);
+ writelock = true;
+ } else {
+ read_lock(&cache->uc_lock);
+ writelock = false;
+ }
+find_with_lock:
list_for_each_entry_safe(entry, next, head, ue_hash) {
/* check invalid & expired items */
- if (check_unlink_entry(cache, entry))
+ if (check_unlink_entry(cache, entry, writelock))
continue;
if (upcall_compare(cache, entry, key, args) == 0) {
found = 1;
if (!found) {
if (!new) {
- spin_unlock(&cache->uc_lock);
+ if (writelock)
+ write_unlock(&cache->uc_lock);
+ else
+ read_unlock(&cache->uc_lock);
new = alloc_entry(cache, key, args);
if (!new) {
- CERROR("fail to alloc entry\n");
+ CERROR("%s: fail to alloc entry: rc = %d\n",
+ cache->uc_name, -ENOMEM);
RETURN(ERR_PTR(-ENOMEM));
}
goto find_again;
if (new) {
free_entry(cache, new);
new = NULL;
+ } else if (!writelock) {
+ /* We found an entry while holding the read lock, so
+ * convert it to a write lock and find again, to check
+ * that entry was not modified/freed in between.
+ */
+ write_lock_from_read(&cache->uc_lock, &writelock);
+ found = 0;
+ goto find_with_lock;
}
list_move(&entry->ue_hash, head);
}
+ /* now we hold a write lock */
get_entry(entry);
/* acquire for new one */
if (UC_CACHE_IS_NEW(entry)) {
UC_CACHE_SET_ACQUIRING(entry);
UC_CACHE_CLEAR_NEW(entry);
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
rc = refresh_entry(cache, entry);
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
entry->ue_acquire_expire = ktime_get_seconds() +
cache->uc_acquire_expire;
if (rc < 0) {
init_wait(&wait);
add_wait_queue(&entry->ue_waitq, &wait);
set_current_state(TASK_INTERRUPTIBLE);
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
left = schedule_timeout(expiry);
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
remove_wait_queue(&entry->ue_waitq, &wait);
if (UC_CACHE_IS_ACQUIRING(entry)) {
/* we're interrupted or upcall failed in the middle */
rc = left > 0 ? -EINTR : -ETIMEDOUT;
+ /* if we waited uc_acquire_expire, we can try again
+ * with same data, but only if acquire is replayable
+ */
+ if (left <= 0 && !cache->uc_acquire_replay)
+ failedacquiring = true;
put_entry(cache, entry);
if (!failedacquiring) {
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
failedacquiring = true;
new = NULL;
CDEBUG(D_OTHER,
entry->ue_key, rc);
goto find_again;
}
- CERROR("acquire for key %llu: error %d\n",
- entry->ue_key, rc);
+ wake_up_all(&entry->ue_waitq);
+ CERROR("%s: acquire for key %lld after %llu: rc = %d\n",
+ cache->uc_name, entry->ue_key,
+ cache->uc_acquire_expire, rc);
GOTO(out, entry = ERR_PTR(rc));
}
}
* We can't refresh the existing one because some
* memory might be shared by multiple processes.
*/
- if (check_unlink_entry(cache, entry)) {
+ if (check_unlink_entry(cache, entry, writelock)) {
/* if expired, try again. but if this entry is
* created by me but too quickly turn to expired
* without any error, should at least give a
* chance to use it once.
*/
if (entry != new) {
+ /* as stated above, we already hold a write lock */
put_entry(cache, entry);
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
new = NULL;
goto find_again;
}
/* Now we know it's good */
out:
- spin_unlock(&cache->uc_lock);
+ if (writelock)
+ write_unlock(&cache->uc_lock);
+ else
+ read_unlock(&cache->uc_lock);
RETURN(entry);
}
EXPORT_SYMBOL(upcall_cache_get_entry);
+void upcall_cache_get_entry_raw(struct upcall_cache_entry *entry)
+{
+ get_entry(entry);
+}
+EXPORT_SYMBOL(upcall_cache_get_entry_raw);
+
+void upcall_cache_update_entry(struct upcall_cache *cache,
+ struct upcall_cache_entry *entry,
+ time64_t expire, int state)
+{
+ write_lock(&cache->uc_lock);
+ entry->ue_expire = expire;
+ if (!state)
+ UC_CACHE_SET_VALID(entry);
+ else
+ entry->ue_flags |= state;
+ write_unlock(&cache->uc_lock);
+}
+EXPORT_SYMBOL(upcall_cache_update_entry);
+
void upcall_cache_put_entry(struct upcall_cache *cache,
struct upcall_cache_entry *entry)
{
}
LASSERT(atomic_read(&entry->ue_refcount) > 0);
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
put_entry(cache, entry);
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
EXIT;
}
EXPORT_SYMBOL(upcall_cache_put_entry);
struct upcall_cache_entry *entry = NULL;
struct list_head *head;
int found = 0, rc = 0;
+ bool writelock = false;
ENTRY;
LASSERT(cache);
head = &cache->uc_hashtable[UC_CACHE_HASH_INDEX(key,
cache->uc_hashsize)];
- spin_lock(&cache->uc_lock);
+ read_lock(&cache->uc_lock);
list_for_each_entry(entry, head, ue_hash) {
if (downcall_compare(cache, entry, key, args) == 0) {
found = 1;
CDEBUG(D_OTHER, "%s: upcall for key %llu not expected\n",
cache->uc_name, key);
/* haven't found, it's possible */
- spin_unlock(&cache->uc_lock);
+ read_unlock(&cache->uc_lock);
RETURN(-EINVAL);
}
if (err) {
CDEBUG(D_OTHER, "%s: upcall for key %llu returned %d\n",
cache->uc_name, entry->ue_key, err);
- GOTO(out, rc = -EINVAL);
+ write_lock_from_read(&cache->uc_lock, &writelock);
+ GOTO(out, rc = err);
}
if (!UC_CACHE_IS_ACQUIRING(entry)) {
CDEBUG(D_RPCTRACE, "%s: found uptodate entry %p (key %llu)"
"\n", cache->uc_name, entry, entry->ue_key);
+ write_lock_from_read(&cache->uc_lock, &writelock);
GOTO(out, rc = 0);
}
if (UC_CACHE_IS_INVALID(entry) || UC_CACHE_IS_EXPIRED(entry)) {
CERROR("%s: found a stale entry %p (key %llu) in ioctl\n",
cache->uc_name, entry, entry->ue_key);
+ write_lock_from_read(&cache->uc_lock, &writelock);
GOTO(out, rc = -EINVAL);
}
- spin_unlock(&cache->uc_lock);
+ read_unlock(&cache->uc_lock);
if (cache->uc_ops->parse_downcall)
rc = cache->uc_ops->parse_downcall(cache, entry, args);
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
if (rc)
GOTO(out, rc);
- entry->ue_expire = ktime_get_seconds() + cache->uc_entry_expire;
+ if (!entry->ue_expire)
+ entry->ue_expire = ktime_get_seconds() + cache->uc_entry_expire;
UC_CACHE_SET_VALID(entry);
CDEBUG(D_OTHER, "%s: created upcall cache entry %p for key %llu\n",
cache->uc_name, entry, entry->ue_key);
out:
+ /* 'goto out' needs to make sure to take a write lock first */
if (rc) {
UC_CACHE_SET_INVALID(entry);
list_del_init(&entry->ue_hash);
}
UC_CACHE_CLEAR_ACQUIRING(entry);
- spin_unlock(&cache->uc_lock);
wake_up(&entry->ue_waitq);
put_entry(cache, entry);
+ write_unlock(&cache->uc_lock);
RETURN(rc);
}
int i;
ENTRY;
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
for (i = 0; i < cache->uc_hashsize; i++) {
list_for_each_entry_safe(entry, next,
&cache->uc_hashtable[i], ue_hash) {
free_entry(cache, entry);
}
}
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
EXIT;
}
EXPORT_SYMBOL(upcall_cache_flush);
head = &cache->uc_hashtable[UC_CACHE_HASH_INDEX(key,
cache->uc_hashsize)];
- spin_lock(&cache->uc_lock);
+ write_lock(&cache->uc_lock);
list_for_each_entry(entry, head, ue_hash) {
if (upcall_compare(cache, entry, key, args) == 0) {
found = 1;
atomic_read(&entry->ue_refcount), entry->ue_flags,
ktime_get_real_seconds(), entry->ue_acquire_expire,
entry->ue_expire);
+ get_entry(entry);
UC_CACHE_SET_EXPIRED(entry);
- if (!atomic_read(&entry->ue_refcount))
- free_entry(cache, entry);
+ put_entry(cache, entry);
}
- spin_unlock(&cache->uc_lock);
+ write_unlock(&cache->uc_lock);
}
EXPORT_SYMBOL(upcall_cache_flush_one);
struct upcall_cache *upcall_cache_init(const char *name, const char *upcall,
- int hashsz, struct upcall_cache_ops *ops)
+ int hashsz, time64_t entry_expire,
+ time64_t acquire_expire, bool replayable,
+ struct upcall_cache_ops *ops)
{
struct upcall_cache *cache;
int i;
if (!cache)
RETURN(ERR_PTR(-ENOMEM));
- spin_lock_init(&cache->uc_lock);
+ rwlock_init(&cache->uc_lock);
init_rwsem(&cache->uc_upcall_rwsem);
cache->uc_hashsize = hashsz;
LIBCFS_ALLOC(cache->uc_hashtable,
RETURN(ERR_PTR(-ENOMEM));
for (i = 0; i < cache->uc_hashsize; i++)
INIT_LIST_HEAD(&cache->uc_hashtable[i]);
- strlcpy(cache->uc_name, name, sizeof(cache->uc_name));
+ strscpy(cache->uc_name, name, sizeof(cache->uc_name));
/* upcall pathname proc tunable */
- strlcpy(cache->uc_upcall, upcall, sizeof(cache->uc_upcall));
- cache->uc_entry_expire = 20 * 60;
- cache->uc_acquire_expire = 30;
+ strscpy(cache->uc_upcall, upcall, sizeof(cache->uc_upcall));
+ cache->uc_entry_expire = entry_expire;
+ cache->uc_acquire_expire = acquire_expire;
+ cache->uc_acquire_replay = replayable;
cache->uc_ops = ops;
RETURN(cache);