if (r > 0)
return; // delayed
+ // Do not batch if any xlock is held
+ if (!r) {
+ CInode *in = mdr->in[0];
+ if (((mask & CEPH_CAP_LINK_SHARED) && (in->linklock.is_xlocked_by_client(client))) ||
+ ((mask & CEPH_CAP_AUTH_SHARED) && (in->authlock.is_xlocked_by_client(client))) ||
+ ((mask & CEPH_CAP_XATTR_SHARED) && (in->xattrlock.is_xlocked_by_client(client))) ||
+ ((mask & CEPH_CAP_FILE_SHARED) && (in->filelock.is_xlocked_by_client(client)))) {
+ r = -1;
+ }
+ }
+
if (r < 0) {
// fall-thru. let rdlock_path_pin_ref() check again.
- } else if (is_lookup) {
+ } else if (is_lookup && mdr->dn[0].size()) {
CDentry* dn = mdr->dn[0].back();
mdr->pin(dn);
auto em = dn->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());