diff --git a/Makefile b/Makefile
index 3b615798..0e1507f5 100644
--- a/Makefile
+++ b/Makefile
@@ -80,11 +80,13 @@ endif
 ifdef FUSE_LL_PATH
 	BIN += bfs_ll_mount
 endif
-TESTS = namespace_test block_mapping_test location_provider_test logdb_test \
+TESTS = namespace_test block_mapping_test \
+		block_mapping_manager_test location_provider_test logdb_test \
 		file_lock_manager_test file_lock_test chunkserver_impl_test \
 	   	file_cache_test block_manager_test data_block_test
 TEST_OBJS = src/nameserver/test/namespace_test.o \
 			src/nameserver/test/block_mapping_test.o \
+			src/nameserver/test/block_mapping_manager_test.o \
 			src/nameserver/test/logdb_test.o \
 			src/nameserver/test/location_provider_test.o \
 			src/nameserver/test/kv_client.o \
@@ -137,6 +139,15 @@ block_mapping_test: src/nameserver/test/block_mapping_test.o src/nameserver/bloc
 	$(CXX) src/nameserver/block_mapping.o src/nameserver/test/block_mapping_test.o \
 	src/nameserver/block_mapping_manager.o $(OBJS) -o $@ $(LDFLAGS)
 
+block_mapping_manager_test: src/nameserver/test/block_mapping_manager_test.o \
+                            src/nameserver/block_mapping_manager.o \
+                            src/nameserver/block_mapping.o
+	$(CXX) src/nameserver/block_mapping.o \
+           src/nameserver/block_mapping_manager.o \
+           src/nameserver/test/block_mapping_manager_test.o \
+           $(OBJS) -o $@ $(LDFLAGS)
+
+
 logdb_test: src/nameserver/test/logdb_test.o src/nameserver/logdb.o
 	$(CXX) src/nameserver/logdb.o src/nameserver/test/logdb_test.o $(OBJS) -o $@ $(LDFLAGS)
 
diff --git a/src/client/bfs_client.cc b/src/client/bfs_client.cc
index 3df73cd3..6341f79b 100644
--- a/src/client/bfs_client.cc
+++ b/src/client/bfs_client.cc
@@ -23,6 +23,7 @@
 
 DECLARE_string(flagfile);
 DECLARE_string(nameserver_nodes);
+DECLARE_int32(sdk_dir_lock_timeout);
 
 void print_usage() {
     printf("Use:\nbfs_client <command> path\n");
@@ -41,6 +42,8 @@ void print_usage() {
     printf("\t    stat : list current stat of the file system\n");
     printf("\t    ln <src> <dst>: create symlink\n");
     printf("\t    chmod <mode> <path> : change file mode bits\n");
+    printf("\t    lockdir <path> : lock the dir\n");
+    printf("\t    unlockdir <path> : unlock the dir\n");
 }
 
 int BfsTouchz(baidu::bfs::FS* fs, int argc, char* argv[]) {
@@ -514,6 +517,14 @@ int BfsShutdownStat(baidu::bfs::FS* fs) {
     return 0;
 }
 
+int BfsLockDir(baidu::bfs::FS* fs, int argc, char* argv[]) {
+    return fs->LockDirectory(argv[0], FLAGS_sdk_dir_lock_timeout);
+}
+
+int BfsUnlockDir(baidu::bfs::FS* fs, int argc, char* argv[]) {
+    return fs->UnlockDirectory(argv[0], true);
+}
+
 /// bfs client main
 int main(int argc, char* argv[]) {
     FLAGS_flagfile = "./bfs.flag";
@@ -569,6 +580,10 @@ int main(int argc, char* argv[]) {
         ret = BfsShutdownStat(fs);
     } else if (strcmp(argv[1], "ln") == 0) {
         ret = BfsLink(fs, argc - 2, argv + 2);
+    } else if (strcmp(argv[1], "lockdir") == 0) {
+        ret = BfsLockDir(fs, argc - 2, argv + 2);
+    } else if (strcmp(argv[1], "unlockdir") == 0) {
+        ret = BfsUnlockDir(fs, argc - 2, argv + 2);
     } else {
         fprintf(stderr, "Unknown command: %s\n", argv[1]);
     }
diff --git a/src/flags.cc b/src/flags.cc
index 77d8102b..46ab476f 100644
--- a/src/flags.cc
+++ b/src/flags.cc
@@ -87,6 +87,7 @@ DEFINE_int32(sdk_thread_num, 10, "Sdk thread num");
 DEFINE_int32(sdk_file_reada_len, 1024*1024, "Read ahead buffer len");
 DEFINE_int32(sdk_createblock_retry, 5, "Create block retry times before fail");
 DEFINE_int32(sdk_write_retry_times, 5, "Write retry times before fail");
+DEFINE_int32(sdk_dir_lock_timeout, 30, "Timeout for get dir lock");
 
 
 /* vim: set expandtab ts=4 sw=4 sts=4 tw=100: */
diff --git a/src/nameserver/block_mapping.cc b/src/nameserver/block_mapping.cc
index 1cdb9e1b..eb8e3ded 100644
--- a/src/nameserver/block_mapping.cc
+++ b/src/nameserver/block_mapping.cc
@@ -969,5 +969,15 @@ void BlockMapping::MarkIncomplete(int64_t block_id) {
     }
 }
 
+RecoverStat BlockMapping::GetRecoverStat(int64_t block_id) {
+    MutexLock lock(&mu_);
+    NSBlock* block = NULL;
+    if (!GetBlockPtr(block_id, &block)) {
+        return kAny;
+    } else {
+        return block->recover_stat;
+    }
+}
+
 } // namespace bfs
 } // namespace baidu
diff --git a/src/nameserver/block_mapping.h b/src/nameserver/block_mapping.h
index b18a1de8..4083f391 100644
--- a/src/nameserver/block_mapping.h
+++ b/src/nameserver/block_mapping.h
@@ -81,6 +81,7 @@ class BlockMapping {
     void ListRecover(RecoverBlockSet* blocks);
     int32_t GetCheckNum();
     void MarkIncomplete(int64_t block_id);
+    RecoverStat GetRecoverStat(int64_t block_id);
 private:
     void DealWithDeadBlockInternal(int32_t cs_id, int64_t block_id);
     typedef std::map<int32_t, std::set<int64_t> > CheckList;
diff --git a/src/nameserver/block_mapping_manager.cc b/src/nameserver/block_mapping_manager.cc
index ebe0e99d..eecd521d 100644
--- a/src/nameserver/block_mapping_manager.cc
+++ b/src/nameserver/block_mapping_manager.cc
@@ -160,6 +160,13 @@ void BlockMappingManager::MarkIncomplete(int64_t block_id) {
 }
 
 bool BlockMappingManager::CheckBlocksClosed(const std::vector<int64_t>& blocks) {
+    for (auto it = blocks.begin(); it != blocks.end(); ++it) {
+        int32_t bucket_offset = GetBucketOffset(*it);
+        RecoverStat stat = block_mapping_[bucket_offset]->GetRecoverStat(*it);
+        if (stat == kBlockWriting || stat == kIncomplete) {
+            return false;
+        }
+    }
     return true;
 }
 
diff --git a/src/nameserver/nameserver_impl.cc b/src/nameserver/nameserver_impl.cc
index cd9e87f3..2fc96da1 100644
--- a/src/nameserver/nameserver_impl.cc
+++ b/src/nameserver/nameserver_impl.cc
@@ -386,7 +386,9 @@ void NameServerImpl::CreateFile(::google::protobuf::RpcController* controller,
     NameServerLog log;
     std::vector<int64_t> blocks_to_remove;
     FileLockGuard file_lock(new WriteLock(path));
-    StatusCode status = namespace_->CreateFile(path, flags, mode, replica_num, &blocks_to_remove, &log);
+    StatusCode status =
+        namespace_->CreateFile(path, flags, mode, replica_num,
+                               &blocks_to_remove, request->uuid(), &log);
     for (size_t i = 0; i < blocks_to_remove.size(); i++) {
         block_mapping_manager_->RemoveBlock(blocks_to_remove[i]);
     }
@@ -781,7 +783,8 @@ void NameServerImpl::Rename(::google::protobuf::RpcController* controller,
     FileInfo remove_file;
     NameServerLog log;
     FileLockGuard file_lock_guard(new WriteLock(oldpath, newpath));
-    StatusCode status = namespace_->Rename(oldpath, newpath, &need_unlink, &remove_file, &log);
+    StatusCode status = namespace_->Rename(oldpath, newpath, &need_unlink,
+                                           &remove_file, request->uuid(), &log);
     response->set_status(status);
     if (status != kOK) {
         done->Run();
@@ -846,7 +849,8 @@ void NameServerImpl::Unlink(::google::protobuf::RpcController* controller,
     FileInfo file_info;
     NameServerLog log;
     FileLockGuard file_lock_guard(new WriteLock(path));
-    StatusCode status = namespace_->RemoveFile(path, &file_info, &log);
+    StatusCode status = namespace_->RemoveFile(path, &file_info,
+                                               request->uuid(), &log);
     sofa::pbrpc::RpcController* ctl = reinterpret_cast<sofa::pbrpc::RpcController*>(controller);
     LOG(INFO, "Sdk %s unlink file %s returns %s",
             ctl->RemoteAddress().c_str(), path.c_str(), StatusCode_Name(status).c_str());
@@ -907,7 +911,8 @@ void NameServerImpl::DeleteDirectory(::google::protobuf::RpcController* controll
     std::vector<FileInfo>* removed = new std::vector<FileInfo>;
     NameServerLog log;
     FileLockGuard file_lock_guard(new WriteLock(path));
-    StatusCode ret_status = namespace_->DeleteDirectory(path, recursive, removed, &log);
+    StatusCode ret_status = namespace_->DeleteDirectory(path, recursive, removed,
+                                                        request->uuid(), &log);
     sofa::pbrpc::RpcController* ctl = reinterpret_cast<sofa::pbrpc::RpcController*>(controller);
     LOG(INFO, "Sdk %s delete directory %s returns %s",
             ctl->RemoteAddress().c_str(), path.c_str(), StatusCode_Name(ret_status).c_str());
@@ -1008,31 +1013,72 @@ void NameServerImpl::LockDir(::google::protobuf::RpcController* controller,
         done->Run();
         return;
     }
+
     std::string path = NameSpace::NormalizePath(request->dir_path());
     FileLockGuard lock_guard(new WriteLock(path));
-    StatusCode status = namespace_->GetDirLockStatus(path);
+    std::string parent_path(path, 0, path.find_last_of("/"));
+    FileInfo info;
+    if (!namespace_->CheckDirLockPermission(parent_path, request->uuid(), &info)) {
+        LOG(INFO, "%s has no permission, parent is locked by %s",
+                request->uuid().c_str(), info.dir_lock_holder_uuid().c_str());
+        response->set_status(kNoPermission);
+        done->Run();
+    }
+
+    std::string holder;
+    StatusCode status = namespace_->GetDirLockStatus(path, &holder);
+    LOG(INFO, "%s try lock dir %s", request->uuid().c_str(), path.c_str());
+    bool need_log_remote = false;
+    NameServerLog log;
     if (status != kDirLocked) {
-        //TODO log remote?
         if (status == kDirUnlock) {
-            namespace_->SetDirLockStatus(path, kDirLocked, request->uuid());
+            namespace_->SetDirLockStatus(path, kDirLocked, request->uuid(), &log);
+            LOG(INFO, "%s lock dir %s", request->uuid().c_str(), path.c_str());
             status = kOK;
+            need_log_remote = true;
         } else if (status == kDirLockCleaning) {
             std::vector<int64_t> blocks;
             namespace_->ListAllBlocks(path, &blocks);
             if (block_mapping_manager_->CheckBlocksClosed(blocks)) {
-                //TODO log remote
-                namespace_->SetDirLockStatus(path, kDirUnlock);
+                namespace_->SetDirLockStatus(path, kDirUnlock,
+                                             request->uuid(), &log);
+                LOG(INFO, "%s lock dir %s",
+                        request->uuid().c_str(), path.c_str());
                 status = kOK;
+                need_log_remote = true;
             }
         } // else status should be kBadParameter
     } else {
-        //TODO log remote
-        namespace_->SetDirLockStatus(path, kDirLockCleaning);
-        status = kDirLockCleaning;
+        if (holder != request->uuid()) {
+            // must set dir lock stat to kDirLockCleaning before ListAllBlocks
+            namespace_->SetDirLockStatus(path, kDirLockCleaning,
+                                         request->uuid(), &log);
+            status = kDirLockCleaning;
+            need_log_remote = true;
+            std::vector<int64_t> blocks;
+            namespace_->ListAllBlocks(path, &blocks);
+            for (auto it = blocks.begin(); it != blocks.end(); ++it) {
+                block_mapping_manager_->MarkIncomplete(*it);
+            }
+            LOG(INFO, "%s try clean %s dir lock",
+                    request->uuid().c_str(), path.c_str());
+        } else {
+            // double lock by the same sdk, ignore it
+            status = kOK;
+        }
     }
+
     response->set_status(status);
-    done->Run();
+    if (need_log_remote) {
+        LogRemote(log, std::bind(&NameServerImpl::SyncLogCallback, this,
+                    controller, request, response, done,
+                    (std::vector<FileInfo>*)NULL, lock_guard,
+                    std::placeholders::_1));
+    } else {
+        done->Run();
+    }
 }
+
 void NameServerImpl::UnlockDir(::google::protobuf::RpcController* controller,
                                const UnlockDirRequest* request,
                                UnlockDirResponse* response,
@@ -1046,19 +1092,29 @@ void NameServerImpl::UnlockDir(::google::protobuf::RpcController* controller,
     FileLockGuard lock_guard(new WriteLock(path));
     StatusCode status = namespace_->GetDirLockStatus(path);
     if (status == kDirLocked) {
-        //TODO log remote
-        namespace_->SetDirLockStatus(path, kDirLockCleaning);
-        std::vector<int64_t> blocks;
-        //TODO add force lock interface which do not list children directory
-        namespace_->ListAllBlocks(path, &blocks);;
-        if (block_mapping_manager_->CheckBlocksClosed(blocks)) {
+        NameServerLog log;
+        if (request->force_unlock()) {
+            namespace_->SetDirLockStatus(path, kDirUnlock, "", &log);
             status = kDirUnlock;
         } else {
-            status = kDirLockCleaning;
+            namespace_->SetDirLockStatus(path, kDirLockCleaning, "", &log);
+            std::vector<int64_t> blocks;
+            namespace_->ListAllBlocks(path, &blocks);;
+            if (block_mapping_manager_->CheckBlocksClosed(blocks)) {
+                status = kDirUnlock;
+            } else {
+                status = kDirLockCleaning;
+            }
         }
+        response->set_status(status);
+        LogRemote(log, std::bind(&NameServerImpl::SyncLogCallback, this,
+                    controller, request, response, done,
+                    (std::vector<FileInfo>*)NULL, lock_guard,
+                    std::placeholders::_1));
+    } else {
+        response->set_status(status);
+        done->Run();
     }
-    response->set_status(status);
-    done->Run();
 }
 
 void NameServerImpl::RebuildBlockMapCallback(const FileInfo& file_info) {
@@ -1615,7 +1671,8 @@ void NameServerImpl::CallMethod(const ::google::protobuf::MethodDescriptor* meth
         std::make_pair("PushBlockReport", work_thread_pool_),
         std::make_pair("SysStat", read_thread_pool_),
         std::make_pair("Chmod", work_thread_pool_),
-        std::make_pair("Symlink", work_thread_pool_)
+        std::make_pair("Symlink", work_thread_pool_),
+        std::make_pair("LockDir", work_thread_pool_)
 
     };
     static int method_num = sizeof(ThreadPoolOfMethod) /
diff --git a/src/nameserver/namespace.cc b/src/nameserver/namespace.cc
index 76c3db04..e1668115 100644
--- a/src/nameserver/namespace.cc
+++ b/src/nameserver/namespace.cc
@@ -251,7 +251,8 @@ bool NameSpace::GetFileInfo(const std::string& path, FileInfo* file_info) {
     return true;
 }
 
-StatusCode NameSpace::BuildPath(const std::string& path, FileInfo* file_info, std::string* fname,
+StatusCode NameSpace::BuildPath(const std::string& path, FileInfo* file_info,
+                                std::string* fname, const std::string& uuid,
                                 NameServerLog* log) {
     std::vector<std::string> paths;
     if (!common::util::SplitPath(path, &paths)) {
@@ -281,6 +282,10 @@ StatusCode NameSpace::BuildPath(const std::string& path, FileInfo* file_info, st
             if (GetFileType(file_info->type()) != kDir) {
                 LOG(INFO, "Create path fail: %s is not a directory", paths[i].c_str());
                 return kBadParameter;
+            } else if (!CheckDirLockPermission(*file_info, uuid)) {
+                LOG(INFO, "Create path %s fail: have no permission on %s",
+                        path.c_str(), paths[i].c_str());
+                return kNoPermission;
             }
         }
         parent_id = file_info->entry_id();
@@ -289,14 +294,17 @@ StatusCode NameSpace::BuildPath(const std::string& path, FileInfo* file_info, st
     return kOK;
 }
 
-StatusCode NameSpace::CreateFile(const std::string& file_name, int flags, int mode, int replica_num,
-                                 std::vector<int64_t>* blocks_to_remove, NameServerLog* log) {
+StatusCode NameSpace::CreateFile(const std::string& file_name, int flags,
+                                 int mode, int replica_num,
+                                 std::vector<int64_t>* blocks_to_remove,
+                                 const std::string& uuid,
+                                 NameServerLog* log) {
     if (file_name == "/") {
         return kBadParameter;
     }
     FileInfo file_info;
     std::string fname, info_value;
-    StatusCode status = BuildPath(file_name, &file_info, &fname, log);
+    StatusCode status = BuildPath(file_name, &file_info, &fname, uuid, log);
     if (status != kOK) {
         return status;
     }
@@ -381,6 +389,7 @@ StatusCode NameSpace::Rename(const std::string& old_path,
                              const std::string& new_path,
                              bool* need_unlink,
                              FileInfo* remove_file,
+                             const std::string& uuid,
                              NameServerLog* log) {
     *need_unlink = false;
     if (old_path == "/" || new_path == "/" || old_path == new_path) {
@@ -391,6 +400,11 @@ StatusCode NameSpace::Rename(const std::string& old_path,
         LOG(INFO, "Rename not found: %s\n", old_path.c_str());
         return kNsNotFound;
     }
+    if (GetFileType(old_file.type()) == kDir &&
+            old_file.dir_lock_stat() == kDirLocked &&
+            old_file.dir_lock_holder_uuid() != uuid) {
+        return kNoPermission;
+    }
 
     std::vector<std::string> new_paths;
     if (!common::util::SplitPath(new_path, &new_paths) || new_paths.empty()) {
@@ -410,6 +424,11 @@ StatusCode NameSpace::Rename(const std::string& old_path,
                 old_path.c_str(), new_path.c_str(), new_paths[i].c_str());
             return kBadParameter;
         }
+        if (!CheckDirLockPermission(path_file, uuid)) {
+            LOG(INFO, "Rename %s to %s fail: have no permission on %s",
+                old_path.c_str(), new_path.c_str(), new_paths[i].c_str());
+            return kNoPermission;
+        }
         if (path_file.entry_id() == old_file.entry_id()) {
             LOG(INFO, "Rename %s to %s fail: %s is the parent directory of %s",
                     old_path.c_str(), new_path.c_str(), old_path.c_str(),
@@ -426,7 +445,8 @@ StatusCode NameSpace::Rename(const std::string& old_path,
         FileInfo dst_file;
         if (LookUp(parent_id, dst_name, &dst_file)) {
             // if dst_file exists, type of both dst_file and old_file must be file
-            if ((GetFileType(dst_file.type()) == kDir) || (GetFileType(old_file.type()) == kDir)) {
+            if ((GetFileType(dst_file.type()) == kDir) ||
+                    (GetFileType(old_file.type()) == kDir)) {
                 LOG(INFO, "Rename %s to %s, src %o or dst %o is not a file",
                     old_path.c_str(), new_path.c_str(), old_file.type(),
                     dst_file.type());
@@ -445,6 +465,7 @@ StatusCode NameSpace::Rename(const std::string& old_path,
     std::string new_key;
     EncodingStoreKey(parent_id, dst_name, &new_key);
     std::string value;
+    // if dst is a directory, the dir lock info will be kept
     old_file.set_parent_entry_id(parent_id);
     old_file.set_name(dst_name);
     old_file.SerializeToString(&value);
@@ -487,7 +508,7 @@ StatusCode NameSpace::Symlink(const std::string& src, const std::string& dst, Na
     }
 
     std::string fname, info_value;
-    StatusCode status = BuildPath(dst, &file_info, &fname, log);
+    StatusCode status = BuildPath(dst, &file_info, &fname, "", log);
     if (status != kOK) {
         return status;
     }
@@ -516,16 +537,29 @@ StatusCode NameSpace::Symlink(const std::string& src, const std::string& dst, Na
     }
 }
 
-StatusCode NameSpace::RemoveFile(const std::string& path, FileInfo* file_removed, NameServerLog* log) {
+StatusCode NameSpace::RemoveFile(const std::string& path,
+                                 FileInfo* file_removed,
+                                 const std::string& uuid,
+                                 NameServerLog* log) {
+    if (path == "/") {
+        return kBadParameter;
+    }
+    std::string parent_path(path, 0, path.find_last_of("/") + 1);
+    FileInfo parent_info;
+    if (!CheckDirLockPermission(parent_path, uuid, &parent_info)) {
+        return kNoPermission;
+    }
+    std::string file_name(path, path.find_last_of("/") + 1,
+                          path.size() - path.find_last_of("/") + 1);
     StatusCode ret_status = kOK;
-    if (LookUp(path, file_removed)) {
+    if (LookUp(parent_info.entry_id(), file_name, file_removed)) {
         // Only support file
         if (GetFileType(file_removed->type()) != kDir) {
             if (path == "/" || path.empty()) {
                 LOG(INFO, "root type= %d", file_removed->type());
             }
             std::string file_key;
-            EncodingStoreKey(file_removed->parent_entry_id(), file_removed->name(), &file_key);
+            EncodingStoreKey(parent_info.entry_id(), file_name, &file_key);
             if (DeleteFileInfo(file_key, log)) {
                 LOG(INFO, "Unlink done: %s\n", path.c_str());
                 ret_status = kOK;
@@ -593,7 +627,9 @@ StatusCode NameSpace::InternalComputeDiskUsage(const FileInfo& info, uint64_t* d
 }
 
 StatusCode NameSpace::DeleteDirectory(const std::string& path, bool recursive,
-                                      std::vector<FileInfo>* files_removed, NameServerLog* log) {
+                                      std::vector<FileInfo>* files_removed,
+                                      const std::string& uuid,
+                                      NameServerLog* log) {
     files_removed->clear();
     FileInfo info;
     if (!LookUp(path, &info)) {
@@ -602,6 +638,9 @@ StatusCode NameSpace::DeleteDirectory(const std::string& path, bool recursive,
     } else if (GetFileType(info.type()) != kDir) {
         LOG(INFO, "Delete Directory, %s %d is not a dir.", path.c_str(), info.type());
         return kBadParameter;
+    } else if (info.dir_lock_stat() == kDirLocked &&
+            info.dir_lock_holder_uuid() != uuid) {
+        return kNoPermission;
     }
     return InternalDeleteDirectory(info, recursive, files_removed, log);
 }
@@ -877,17 +916,156 @@ int64_t NameSpace::GetNewBlockId() {
     return next_block_id_++;
 }
 
-StatusCode NameSpace::GetDirLockStatus(const std::string& path) {
-    return kDirUnlock;
+StatusCode NameSpace::GetDirLockStatus(const std::string& path,
+                                       std::string* holder) {
+    FileInfo info;
+    if (!LookUp(path, &info)) {
+        return kNsNotFound;
+    } else if (GetFileType(info.type()) != kDir) {
+        return kBadParameter;
+    } else {
+        if (holder && info.dir_lock_stat() == kDirLocked) {
+            *holder = info.dir_lock_holder_uuid();
+        }
+        return info.dir_lock_stat();
+    }
 }
 
-void NameSpace::SetDirLockStatus(const std::string& path, StatusCode status,
-                                 const std::string& uuid) {
-
+StatusCode NameSpace::SetDirLockStatus(const std::string& path,
+                                       StatusCode status,
+                                       const std::string& uuid,
+                                       NameServerLog* log) {
+    std::vector<std::string> paths;
+    common::util::SplitPath(path, &paths);
+    int64_t entry_id = kRootEntryid;
+    std::string file_key;
+    FileInfo info;
+    for (size_t i = 0; i < paths.size() - 1; i++) {
+        EncodingStoreKey(entry_id, paths[i], &file_key);
+        bool r = GetFromStore(file_key, &info);
+        if (!r) {
+            return kNsNotFound;
+        }
+        // all the parent directories should be clear
+        if (info.dir_lock_stat() == kDirLocked) {
+            return kNoPermission;
+        }
+        entry_id = info.entry_id();
+    }
+    EncodingStoreKey(entry_id, paths[paths.size() - 1], &file_key);
+    bool r = GetFromStore(file_key, &info);
+    if (!r) {
+        return kNsNotFound;
+    }
+    info.set_dir_lock_stat(status);
+    if (status == kDirUnlock) {
+        info.set_dir_lock_holder_uuid("");
+    } else if (status == kDirLocked) {
+        info.set_dir_lock_holder_uuid(uuid);
+    }
+    std::string info_buf;
+    info.SerializeToString(&info_buf);
+    db_->Put(leveldb::WriteOptions(), file_key, info_buf);
+    EncodeLog(log, kSyncWrite, file_key, info_buf);
+    return kOK;
 }
 
 void NameSpace::ListAllBlocks(const std::string& path, std::vector<int64_t>* result) {
+    //TODO modify return value
+    FileInfo info;
+    if (!LookUp(path, &info)) {
+        return;
+    }
+    if (GetFileType(info.type()) == kDefault) {
+        for (int i = 0; i < info.blocks_size(); i++) {
+            result->push_back(info.blocks(i));
+        }
+        return;
+    }
+    ListAllBlocks(info.entry_id(), result);
+}
 
+void NameSpace::ListAllBlocks(int64_t entry_id, std::vector<int64_t>* result) {
+    std::string key_start, key_end;
+    EncodingStoreKey(entry_id, "", &key_start);
+    EncodingStoreKey(entry_id + 1, "", &key_end);
+    leveldb::Iterator* it = db_->NewIterator(leveldb::ReadOptions());
+    for (it->Seek(key_start); it->Valid(); it->Next()) {
+        leveldb::Slice key = it->key();
+        if (key.compare(key_end)>=0) {
+            break;
+        }
+        FileInfo info;
+        GetFromStore(std::string(key.data(), key.size()), &info);
+        FileType type = GetFileType(info.type());
+        if (type == kDefault) {
+            for (int i = 0; i < info.blocks_size(); i++) {
+                result->push_back(info.blocks(i));
+            }
+        } else if (type == kDir){
+            ListAllBlocks(info.entry_id(), result);
+        }
+    }
+    delete it;
+}
+
+bool NameSpace::CheckDirLockPermission(const std::string& path,
+                                       const std::string& uuid,
+                                       FileInfo* info) {
+    if (path == "/") {
+        StatusCode status = root_path_.dir_lock_stat();
+        if (status == kDirLocked && uuid != root_path_.dir_lock_holder_uuid()) {
+            return false;
+        } else if (status == kDirLockCleaning) {
+            return false;
+        }
+        *info = root_path_;
+        return true;
+    }
+    std::vector<std::string> paths;
+    common::util::SplitPath(path, &paths);
+    int64_t entry_id = kRootEntryid;
+    for (size_t i = 0; i < paths.size(); i++) {
+        std::string file_key;
+        EncodingStoreKey(entry_id, paths[i], &file_key);
+        if(!GetFromStore(file_key, info)) {
+            return false;
+        } else {
+            if (GetFileType(info->type()) == kDir) {
+                StatusCode status = info->dir_lock_stat();
+                if (status == kDirLocked &&
+                        info->dir_lock_holder_uuid() != uuid) {
+                    LOG(INFO, "No permission, %s is lock by %s",
+                            paths[i].c_str(),
+                            info->dir_lock_holder_uuid().c_str());
+                    return false;
+                } else if (status == kDirLockCleaning) {
+                    LOG(INFO, "No permission, %s dir lock is being cleaning",
+                            paths[i].c_str());
+                    return false;
+                }
+            }
+            entry_id = info->entry_id();
+        }
+    }
+    return true;
+}
+
+bool NameSpace::CheckDirLockPermission(const FileInfo& file_info,
+                                       const std::string& uuid) {
+    StatusCode status = file_info.dir_lock_stat();
+    if (status == kDirLocked && file_info.dir_lock_holder_uuid() != uuid) {
+        LOG(INFO, "%s no permission, %s is locked by %s",
+                uuid.c_str(), file_info.name().c_str(),
+                file_info.dir_lock_holder_uuid().c_str());
+        return false;
+    } else if (status == kDirLockCleaning &&
+            file_info.dir_lock_holder_uuid() == uuid) {
+        LOG(INFO, "%s no permission, %s dir lock is being cleaning",
+                uuid.c_str(), file_info.name().c_str());
+        return false;
+    }
+    return true;
 }
 
 } // namespace bfs
diff --git a/src/nameserver/namespace.h b/src/nameserver/namespace.h
index 42e46e87..c959f046 100644
--- a/src/nameserver/namespace.h
+++ b/src/nameserver/namespace.h
@@ -32,19 +32,25 @@ class NameSpace {
                       google::protobuf::RepeatedPtrField<FileInfo>* outputs);
     /// Create file by name
     StatusCode CreateFile(const std::string& file_name, int flags, int mode,
-                          int replica_num, std::vector<int64_t>* blocks_to_remove,
+                          int replica_num,
+                          std::vector<int64_t>* blocks_to_remove,
+                          const std::string& uuid = "",
                           NameServerLog* log = NULL);
     /// Remove file by name
-    StatusCode RemoveFile(const std::string& path, FileInfo* file_removed, NameServerLog* log = NULL);
+    StatusCode RemoveFile(const std::string& path, FileInfo* file_removed,
+                          const std::string& uuid = "", NameServerLog* log = NULL);
     /// Remove director.
     StatusCode DeleteDirectory(const std::string& path, bool recursive,
-                               std::vector<FileInfo>* files_removed, NameServerLog* log = NULL);
+                               std::vector<FileInfo>* files_removed,
+                               const std::string& uuid = "",
+                               NameServerLog* log = NULL);
     StatusCode DiskUsage(const std::string& path, uint64_t* du_size);
     /// File rename
     StatusCode Rename(const std::string& old_path,
                       const std::string& new_path,
                       bool* need_unlink,
                       FileInfo* remove_file,
+                      const std::string& uuid = "",
                       NameServerLog* log = NULL);
     /// Symlink: dst -> src
     StatusCode Symlink(const std::string& src,
@@ -68,10 +74,15 @@ class NameSpace {
     void TailSnapshot(int32_t ns_id, std::string* logstr);
     void EraseNamespace();
     int64_t GetNewBlockId();
-    StatusCode GetDirLockStatus(const std::string& path);
-    void SetDirLockStatus(const std::string& path, StatusCode status,
-                          const std::string& uuid = "");
+    StatusCode GetDirLockStatus(const std::string& path,
+                                std::string* holder = NULL);
+    StatusCode SetDirLockStatus(const std::string& path, StatusCode status,
+                                const std::string& uuid = "",
+                                NameServerLog* log = NULL);
     void ListAllBlocks(const std::string& path, std::vector<int64_t>* result);
+    bool CheckDirLockPermission(const std::string& path,
+                                const std::string& uuid,
+                                FileInfo* info);
 private:
     enum FileType {
         kDefault = 0,
@@ -80,8 +91,9 @@ class NameSpace {
     };
     FileType GetFileType(int type) const;
     bool GetLinkSrcPath(const FileInfo& info, FileInfo* src_info);
-    StatusCode BuildPath(const std::string& path, FileInfo* file_info, std::string* fname,
-                                NameServerLog* log = NULL);
+    StatusCode BuildPath(const std::string& path, FileInfo* file_info,
+                         std::string* fname, const std::string& uuid = "",
+                         NameServerLog* log = NULL);
     static void EncodingStoreKey(int64_t entry_id,
                           const std::string& path,
                           std::string* key_str);
@@ -101,6 +113,8 @@ class NameSpace {
                        const std::string& key, const std::string& value);
     void InitBlockIdUpbound(NameServerLog* log);
     void UpdateBlockIdUpbound(NameServerLog* log);
+    void ListAllBlocks(int64_t entry_id, std::vector<int64_t>* result);
+    bool CheckDirLockPermission(const FileInfo& info, const std::string& uuid);
 private:
     leveldb::DB* db_;   /// NameSpace storage
     leveldb::Cache* db_cache_;  // block cache for leveldb
diff --git a/src/nameserver/test/block_mapping_manager_test.cc b/src/nameserver/test/block_mapping_manager_test.cc
new file mode 100644
index 00000000..d2cda672
--- /dev/null
+++ b/src/nameserver/test/block_mapping_manager_test.cc
@@ -0,0 +1,59 @@
+// Copyright (c) 2017, Baidu.com, Inc. All Rights Reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+//
+
+#define private public
+#include "nameserver/block_mapping_manager.h"
+
+#include <gtest/gtest.h>
+
+namespace baidu {
+namespace bfs {
+
+class BlockMappingManagerTest : public ::testing::Test {
+public:
+    BlockMappingManagerTest() {}
+protected:
+};
+
+TEST_F(BlockMappingManagerTest, CheckBlocksClosed) {
+    BlockMappingManager bm(5);
+    int64_t block_id1 = 1;
+    int64_t block_id2 = 2;
+    int64_t block_version = 0;
+    int64_t block_size = 0;
+    int32_t replica = 3;
+    bm.RebuildBlock(block_id1, replica, block_version, block_size);
+    bm.RebuildBlock(block_id2, replica, block_version, block_size);
+    std::vector<int64_t> blocks;
+    blocks.push_back(block_id1);
+    blocks.push_back(block_id2);
+    bool ret = bm.CheckBlocksClosed(blocks);
+    int32_t cs1 = 23;
+    int32_t cs2 = 45;
+    int32_t cs3 = 67;
+    ret =
+        bm.UpdateBlockInfo(block_id1, cs1, block_size, block_version) &&
+        bm.UpdateBlockInfo(block_id1, cs2, block_size, block_version) &&
+        bm.UpdateBlockInfo(block_id1, cs3, block_size, block_version);
+    ASSERT_EQ(ret, true);
+    ret = bm.CheckBlocksClosed(blocks);
+    ASSERT_EQ(ret, false);
+    ret =
+        bm.UpdateBlockInfo(block_id2, cs1, block_size, block_version) &&
+        bm.UpdateBlockInfo(block_id2, cs2, block_size, block_version) &&
+        bm.UpdateBlockInfo(block_id2, cs3, block_size, block_version);
+    ASSERT_EQ(ret, true);
+    ret = bm.CheckBlocksClosed(blocks);
+    ASSERT_EQ(ret, true);
+}
+
+} // namespace bfs
+} // namespace baidu
+
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}
+
diff --git a/src/nameserver/test/block_mapping_test.cc b/src/nameserver/test/block_mapping_test.cc
index d2deda8f..77ef2432 100644
--- a/src/nameserver/test/block_mapping_test.cc
+++ b/src/nameserver/test/block_mapping_test.cc
@@ -126,6 +126,27 @@ TEST_F(BlockMappingTest, NotRecoverEmptyBlock) {
     ASSERT_TRUE(bm->lost_blocks_.empty());
 }
 
+TEST_F(BlockMappingTest, GetRecoverStat) {
+    int64_t block_id = 1;
+    int64_t block_version = 0;
+    int64_t block_size = 0;
+    int32_t replica = 3;
+    BlockMapping* bm = new BlockMapping(&thread_pool);
+    bm->RebuildBlock(block_id, replica, block_version, block_size);
+    int32_t cs1 = 23;
+    int32_t cs2 = 45;
+    int32_t cs3 = 67;
+    bool ret =
+        bm->UpdateBlockInfo(block_id, cs1, block_size, block_version) &&
+        bm->UpdateBlockInfo(block_id, cs2, block_size, block_version);
+    ASSERT_TRUE(ret);
+    RecoverStat stat = bm->GetRecoverStat(block_id);
+    ASSERT_EQ(stat, kBlockWriting);
+    bm->UpdateBlockInfo(block_id, cs3, block_size, block_version);
+    stat = bm->GetRecoverStat(block_id);
+    ASSERT_EQ(stat, kNotInRecover);
+}
+
 } // namespace bfs
 } // namespace baidu
 
diff --git a/src/nameserver/test/namespace_test.cc b/src/nameserver/test/namespace_test.cc
index 8efaa9e7..701ee193 100644
--- a/src/nameserver/test/namespace_test.cc
+++ b/src/nameserver/test/namespace_test.cc
@@ -133,6 +133,14 @@ TEST_F(NameSpaceTest, CreateFile) {
     ASSERT_EQ(kOK, ns.CreateFile("/dir1/subdir2/file3", 0, 01755, -1, &blocks_to_remove));
     ASSERT_EQ(kBadParameter, ns.CreateFile("/", 0, 01755, -1, &blocks_to_remove));
     ASSERT_EQ(kBadParameter, ns.CreateFile("/", 0, 0, -1, &blocks_to_remove));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir1/subdir1", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.CreateFile("/dir1/subdir1/file4", 0, 0, -1, &blocks_to_remove, "uuid1"));
+    ASSERT_EQ(kOK, ns.CreateFile("/dir1/subdir1/file4", 0, 0, -1, &blocks_to_remove, "uuid0"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir1/subdir1", kDirLockCleaning, ""));
+    ASSERT_EQ(kNoPermission, ns.CreateFile("/dir1/subdir1/file5", 0, 0, -1, &blocks_to_remove, "uuid0"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir1/subdir1", kDirUnlock, ""));
+    ASSERT_EQ(kOK, ns.CreateFile("/dir1/subdir1/file5", 0, 0, -1, &blocks_to_remove, "uuid0"));
+    system("rm -rf ./db");
 }
 
 TEST_F(NameSpaceTest, List) {
@@ -152,7 +160,10 @@ TEST_F(NameSpaceTest, List) {
 }
 
 TEST_F(NameSpaceTest, Symlink) {
+    FLAGS_namedb_path = "./db";
+    system("rm -rf ./db");
     NameSpace ns;
+    ASSERT_TRUE(CreateTree(&ns));
     std::vector<int64_t> blocks_to_remove;
     /// link -> file
     ASSERT_EQ(kOK, ns.Symlink("/dir1/subdir1/file3", "/link3"));
@@ -166,10 +177,14 @@ TEST_F(NameSpaceTest, Symlink) {
     ASSERT_EQ(kFileExists, ns.Symlink("/file1", "/file1"));
     /// none -> link
     ASSERT_EQ(kNsNotFound, ns.Symlink("/file000", "/link5"));
+    system("rm -rf ./db");
 }
 
 TEST_F(NameSpaceTest, Rename) {
+    FLAGS_namedb_path = "./db";
+    system("rm -rf ./db");
     NameSpace ns;
+    ASSERT_TRUE(CreateTree(&ns));
     bool need_unlink;
     FileInfo remove_file;
     /// self -> self
@@ -222,12 +237,42 @@ TEST_F(NameSpaceTest, Rename) {
 
 
     ///  link A -> link B
-    ASSERT_EQ(kOK, ns.Rename("/link4", "/link", &need_unlink, &remove_file));
+    ASSERT_EQ(kOK, ns.Rename("/link2", "/link4", &need_unlink, &remove_file));
     ASSERT_FALSE(need_unlink);
     /// fileA -> fileB, B is link
     ASSERT_EQ(kOK, ns.Rename("/file2", "/link", &need_unlink, &remove_file));
     ASSERT_FALSE(need_unlink);
 
+    /// rename dir which parent protected by dir lock
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/home", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.Rename("/home/dir2", "/home/dir2_lock",
+                                       &need_unlink, &remove_file, "uuid1"));
+    ASSERT_EQ(kOK, ns.Rename("/home/dir2", "/home/dir2_lock",
+                             &need_unlink, &remove_file, "uuid0"));
+
+    /// rename dir after dir lock removed
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/home", kDirLockCleaning, ""));
+    // not allow self's writing when cleaning dir lock
+    ASSERT_EQ(kNoPermission, ns.Rename("/home/dir2_lock", "/home/dir2",
+                             &need_unlink, &remove_file, "uuid0"));
+    // allow other's writing when cleaning dir lock
+    ASSERT_EQ(kOK, ns.Rename("/home/dir2_lock", "/home/dir2",
+                             &need_unlink, &remove_file, "uuid1"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/home", kDirUnlock, ""));
+    ASSERT_EQ(kOK, ns.Rename("/home/dir2", "/home/dir2_lock",
+                             &need_unlink, &remove_file, "uuid1"));
+
+    /// rename dir protected by dir lock
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/home", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.Rename("/home", "/home1",
+                             &need_unlink, &remove_file, "uuid1"));
+    ASSERT_EQ(kOK, ns.Rename("/home", "/home1",
+                             &need_unlink, &remove_file, "uuid0"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/home1", kDirLockCleaning, ""));
+    ASSERT_EQ(kOK, ns.Rename("/home1", "/home",
+                             &need_unlink, &remove_file, "uuid0"));
+    ASSERT_EQ(kDirLockCleaning, ns.GetDirLockStatus("/home"));
+    system("rm -rf ./db");
 }
 
 TEST_F(NameSpaceTest, RemoveFile) {
@@ -250,6 +295,19 @@ TEST_F(NameSpaceTest, RemoveFile) {
     ASSERT_EQ(kOK, ns.RemoveFile("/link1", &file_removed));
     ASSERT_EQ(11, file_removed.entry_id());
 
+    /// rm file protected by dir lock
+    std::vector<int64_t> blocks_to_remove;
+    ASSERT_EQ(kOK, ns.CreateFile("/dir/subdir1/file3", 0, 0, -1, &blocks_to_remove));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir/subdir1", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.RemoveFile("/dir/subdir1/file3", &file_removed, "uuid1"));
+    ASSERT_EQ(kOK, ns.RemoveFile("/dir/subdir1/file3", &file_removed, "uuid0"));
+    ASSERT_EQ(kOK, ns.CreateFile("/dir/subdir1/file4", 0, 0, -1, &blocks_to_remove, "uuid0"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir/subdir1", kDirLockCleaning, ""));
+    ASSERT_EQ(kNoPermission, ns.RemoveFile("/dir/subdir1/file4", &file_removed, "uuid1"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/dir/subdir1", kDirUnlock, ""));
+    ASSERT_EQ(kOK, ns.RemoveFile("/dir/subdir1/file4", &file_removed, "uuid1"));
+
+    system("rm -rf ./db");
 }
 
 TEST_F(NameSpaceTest, DeleteDirectory) {
@@ -295,6 +353,22 @@ TEST_F(NameSpaceTest, DeleteDirectory) {
     ASSERT_EQ(kOK, ns.DeleteDirectory("/", true, &files_removed));
     ASSERT_EQ(kOK, ns.ListDirectory("/", &outputs));
     ASSERT_EQ(0, outputs.size());
+
+    // Rmr dir with dir lock protected
+    std::vector<int64_t> blocks_to_remove;
+    ns.CreateFile("/tera", 0, 01755, -1, &blocks_to_remove);
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/tera", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.DeleteDirectory("/tera", true, &files_removed, "uuid1"));
+    ASSERT_EQ(kOK, ns.ListDirectory("/", &outputs));
+    ASSERT_EQ(1, outputs.size());
+    ASSERT_EQ(kOK, ns.DeleteDirectory("/tera", true, &files_removed, "uuid0"));
+    ASSERT_EQ(kOK, ns.ListDirectory("/", &outputs));
+    ASSERT_EQ(0, outputs.size());
+    ns.CreateFile("/tera", 0, 01755, -1, &blocks_to_remove);
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/tera", kDirLocked, "uuid0"));
+    ASSERT_EQ(kNoPermission, ns.DeleteDirectory("/tera", true, &files_removed, "uuid1"));
+    ASSERT_EQ(kOK, ns.SetDirLockStatus("/tera", kDirLockCleaning, ""));
+    ASSERT_EQ(kOK, ns.DeleteDirectory("/tera", true, &files_removed, "uuid1"));
 }
 
 TEST_F(NameSpaceTest, DeleteDirectory2) {
@@ -363,6 +437,110 @@ TEST_F(NameSpaceTest, GetNewBlockId) {
     system("rm -rf ./db");
 }
 
+TEST_F(NameSpaceTest, ListAllBlocks) {
+    system("rm -rf ./db");
+    std::vector<int64_t> blocks_to_remove;
+    NameSpace ns;
+    ns.CreateFile("/abc", 0, 01755, -1, &blocks_to_remove);
+    ns.CreateFile("/abc/def", 0, 0, -1, &blocks_to_remove);
+    ns.CreateFile("/abc/ghi", 0, 0, -1, &blocks_to_remove);
+    FileInfo info;
+    ns.GetFileInfo("/abc/def", &info);
+    ASSERT_EQ(info.blocks_size(), 0);
+    info.add_blocks(ns.GetNewBlockId());
+    ns.UpdateFileInfo(info, NULL);
+    ns.GetFileInfo("/abc/ghi", &info);
+    info.add_blocks(ns.GetNewBlockId());
+    ns.UpdateFileInfo(info, NULL);
+    std::vector<int64_t> all_blocks;
+    ns.ListAllBlocks("/abc", &all_blocks);
+    ASSERT_EQ(all_blocks.size(), 2);
+    ns.CreateFile("/abc/jkl/mno", 0, 0, -1, &blocks_to_remove);
+    ns.GetFileInfo("abc/jkl/mno", &info);
+    info.add_blocks(ns.GetNewBlockId());
+    ns.UpdateFileInfo(info, NULL);
+    all_blocks.clear();
+    ns.ListAllBlocks("/abc", &all_blocks);
+    ASSERT_EQ(all_blocks.size(), 3);
+    all_blocks.clear();
+    ns.ListAllBlocks("/", &all_blocks);
+    ASSERT_EQ(all_blocks.size(), 3);
+    all_blocks.clear();
+    ns.ListAllBlocks("/abc/def", &all_blocks);
+    ASSERT_EQ(all_blocks.size(), 1);
+    all_blocks.clear();
+    ns.GetFileInfo("/abc/def", &info);
+    info.add_blocks(ns.GetNewBlockId());
+    ns.UpdateFileInfo(info, NULL);
+    ns.ListAllBlocks("/abc/def", &all_blocks);
+    ASSERT_EQ(all_blocks.size(), 2);
+    system("rm -rf ./db");
+}
+
+TEST_F(NameSpaceTest, GetAndSetDirLockStatus) {
+    system("rm -rf ./db");
+    NameSpace ns;
+    std::vector<int64_t> blocks_to_remove;
+    std::string path0("/abc");
+    ns.CreateFile(path0, 0, 01755, -1, &blocks_to_remove);
+    StatusCode status = ns.GetDirLockStatus(path0);
+    ASSERT_EQ(status, kDirUnlock);
+    status = ns.SetDirLockStatus(path0, kDirLocked);
+    ASSERT_EQ(status, kOK);
+    status = ns.GetDirLockStatus(path0);
+    ASSERT_EQ(status, kDirLocked);
+
+    std::string path1("/def");
+    status = ns.GetDirLockStatus(path1);
+    ASSERT_EQ(status, kNsNotFound);
+
+    std::string path2("/abc/ghi");
+    ns.CreateFile(path2, 0, 01755, -1, &blocks_to_remove);
+    status = ns.GetDirLockStatus(path2);
+    ASSERT_EQ(status, kDirLocked);
+    status = ns.SetDirLockStatus(path2, kDirLocked);
+    ASSERT_EQ(status, kNoPermission);
+
+    status = ns.SetDirLockStatus(path0, kDirLockCleaning);
+    ASSERT_EQ(status, kOK);
+    status = ns.GetDirLockStatus(path0);
+    ASSERT_EQ(status, kDirLockCleaning);
+
+    system("rm -rf ./db");
+}
+
+TEST_F(NameSpaceTest, CheckDirLockPermission) {
+    system("rm -rf ./db");
+    NameSpace ns;
+    std::vector<int64_t> blocks_to_remove;
+    FileInfo info;
+    ns.CreateFile("/abc/def/ghi", 0, 01755, -1, &blocks_to_remove);
+    ASSERT_TRUE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid0", &info));
+    ASSERT_TRUE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid1", &info));
+    ns.SetDirLockStatus("/abc/def/ghi", kDirLocked, "uuid0");
+    ASSERT_FALSE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid1", &info));
+    ASSERT_TRUE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid0", &info));
+    ns.SetDirLockStatus("/abc/def/ghi", kDirLockCleaning, "");
+    ASSERT_FALSE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid1", &info));
+    ASSERT_FALSE(ns.CheckDirLockPermission("/abc/def/ghi", "uuid0", &info));
+
+    ns.SetDirLockStatus("/abc/def/ghi", kDirUnlock, "");
+    ASSERT_TRUE(ns.LookUp("/abc/def/ghi", &info));
+    ASSERT_TRUE(ns.CheckDirLockPermission(info, "uuid0"));
+    ASSERT_TRUE(ns.CheckDirLockPermission(info, "uuid1"));
+
+    ns.SetDirLockStatus("/abc/def/ghi", kDirLocked, "uuid0");
+    ASSERT_TRUE(ns.LookUp("/abc/def/ghi", &info));
+    ASSERT_TRUE(ns.CheckDirLockPermission(info, "uuid0"));
+    ASSERT_FALSE(ns.CheckDirLockPermission(info, "uuid1"));
+
+    ns.SetDirLockStatus("/abc/def/ghi", kDirLockCleaning, "uuid0");
+    ASSERT_TRUE(ns.LookUp("/abc/def/ghi", &info));
+    ASSERT_FALSE(ns.CheckDirLockPermission(info, "uuid0"));
+    ASSERT_TRUE(ns.CheckDirLockPermission(info, "uuid1"));
+    system("rm -rf ./db");
+}
+
 }
 }
 
diff --git a/src/proto/file.proto b/src/proto/file.proto
index c8205358..1f523d25 100644
--- a/src/proto/file.proto
+++ b/src/proto/file.proto
@@ -1,3 +1,5 @@
+import "status_code.proto";
+
 package baidu.bfs;
 
 message FileInfo {
@@ -13,5 +15,7 @@ message FileInfo {
     optional int32 owner = 10;
     repeated string cs_addrs = 11;
     optional string sym_link = 12;
+    optional StatusCode dir_lock_stat = 13 [default = kDirUnlock];
+    optional string dir_lock_holder_uuid = 14;
 }
 
diff --git a/src/proto/nameserver.proto b/src/proto/nameserver.proto
index b867d127..9b32ca18 100644
--- a/src/proto/nameserver.proto
+++ b/src/proto/nameserver.proto
@@ -42,6 +42,7 @@ message CreateFileRequest {
     optional int32 flags = 4;
     optional int32 replica_num = 5;
     optional string user = 7;
+    optional string uuid = 8;
 }
 
 message CreateFileResponse {
@@ -94,6 +95,7 @@ message RenameRequest {
     optional int64 sequence_id = 1;
     optional string oldpath = 2;
     optional string newpath = 3;
+    optional string uuid = 4;
 }
 message RenameResponse {
     optional int64 sequence_id = 1;
@@ -139,6 +141,7 @@ message FinishBlockResponse {
 message UnlinkRequest {
     optional int64 sequence_id = 1;
     optional string path = 2;
+    optional string uuid = 3;
 }
 message UnlinkResponse {
     optional int64 sequence_id = 1;
@@ -149,6 +152,7 @@ message DeleteDirectoryRequest {
     optional int64 sequence_id = 1;
     optional string path = 2;
     optional bool recursive = 3;
+    optional string uuid = 4;
 }
 message DeleteDirectoryResponse {
     optional int64 sequence_id = 1;
@@ -353,6 +357,7 @@ message UnlockDirRequest {
     optional int64 sequence_id = 1;
     optional string dir_path = 2;
     optional string uuid = 3;
+    optional bool force_unlock = 4 [default = false];
 };
 
 message UnlockDirResponse {
diff --git a/src/sdk/bfs.h b/src/sdk/bfs.h
index e8a534f9..07bbe7c2 100644
--- a/src/sdk/bfs.h
+++ b/src/sdk/bfs.h
@@ -99,9 +99,9 @@ class FS {
     /// Delete Directory
     virtual int32_t DeleteDirectory(const char* path, bool recursive) = 0;
     /// Lock Directory
-    virtual int32_t LockDirectory(const char* path) = 0;
+    virtual int32_t LockDirectory(const char* path, int32_t timeout = -1) = 0;
     /// Unlock Directory
-    virtual int32_t UnlockDirectory(const char* path) = 0;
+    virtual int32_t UnlockDirectory(const char* path, bool force_unloak = false) = 0;
     /// Du
     virtual int32_t DiskUsage(const char* path, int64_t* du_size) = 0;
     /// Access
diff --git a/src/sdk/fs_impl.cc b/src/sdk/fs_impl.cc
index 473567f6..dce9c390 100644
--- a/src/sdk/fs_impl.cc
+++ b/src/sdk/fs_impl.cc
@@ -100,6 +100,7 @@ int32_t FSImpl::CreateDirectory(const char* path) {
     request.set_file_name(path);
     request.set_mode(0755|(1<<9));
     request.set_sequence_id(0);
+    request.set_uuid(GetUUID());
     bool ret = nameserver_client_->SendRequest(&NameServer_Stub::CreateFile,
         &request, &response, 15, 3);
     if (!ret) {
@@ -165,6 +166,7 @@ int32_t FSImpl::DeleteDirectory(const char* path, bool recursive) {
     request.set_sequence_id(0);
     request.set_path(path);
     request.set_recursive(recursive);
+    request.set_uuid(GetUUID());
     bool ret = nameserver_client_->SendRequest(&NameServer_Stub::DeleteDirectory,
             &request, &response, 3600, 1);
     if (!ret) {
@@ -345,6 +347,7 @@ int32_t FSImpl::OpenFile(const char* path, int32_t flags, int32_t mode,
     request.set_flags(flags);
     request.set_mode(mode&0777);
     request.set_replica_num(write_option.replica);
+    request.set_uuid(GetUUID());
     bool rpc_ret = nameserver_client_->SendRequest(&NameServer_Stub::CreateFile,
         &request, &response, 15, 1);
     if (!rpc_ret || response.status() != kOK) {
@@ -397,6 +400,7 @@ int32_t FSImpl::DeleteFile(const char* path) {
     request.set_path(path);
     int64_t seq = common::timer::get_micros();
     request.set_sequence_id(seq);
+    request.set_uuid(GetUUID());
     // printf("Delete file: %s\n", path);
     bool ret = nameserver_client_->SendRequest(&NameServer_Stub::Unlink,
         &request, &response, 15, 1);
@@ -410,24 +414,35 @@ int32_t FSImpl::DeleteFile(const char* path) {
     }
     return OK;
 }
-int32_t FSImpl::LockDirectory(const char* path) {
-    //TODO Support set timeout for LockDirectory
+int32_t FSImpl::LockDirectory(const char* path, int32_t timeout) {
     LockDirRequest request;
     LockDirResponse response;
     request.set_dir_path(path);
     request.set_uuid(GetUUID());
+    int32_t start_lock_time = common::timer::now_time();
+    bool is_timeout = false;
     while (!nameserver_client_->SendRequest(&NameServer_Stub::LockDir,
                 &request, &response, 15, 1) || response.status() != kOK) {
-        sleep(5);
+        if (common::timer::now_time() > timeout + start_lock_time) {
+            LOG(INFO, "Get %s dir lock timeout", path);
+            is_timeout = true;
+            break;
+        } else {
+            sleep(5);
+        }
+    }
+    if (is_timeout) {
+        return TIMEOUT;
+    } else {
+        return OK;
     }
-    assert(response.status() == kOK);
-    return OK;
 }
-int32_t FSImpl::UnlockDirectory(const char* path) {
+int32_t FSImpl::UnlockDirectory(const char* path, bool force_unloak) {
     UnlockDirRequest request;
     UnlockDirResponse response;
     request.set_dir_path(path);
     request.set_uuid(GetUUID());
+    request.set_force_unlock(force_unloak);
     nameserver_client_->SendRequest(&NameServer_Stub::UnlockDir,
             &request, &response, 15, 1);
     //Don't care return value of rpc
@@ -439,6 +454,7 @@ int32_t FSImpl::Rename(const char* oldpath, const char* newpath) {
     request.set_oldpath(oldpath);
     request.set_newpath(newpath);
     request.set_sequence_id(0);
+    request.set_uuid(GetUUID());
     bool ret = nameserver_client_->SendRequest(&NameServer_Stub::Rename,
         &request, &response, 15, 1);
     if (!ret) {
@@ -584,7 +600,9 @@ bool FS::OpenFileSystem(const char* nameserver, FS** fs, const FSOptions&) {
 }
 
 const std::string& FSImpl::GetUUID() {
-    static std::string uuid;
+    static std::string uuid = common::util::GetLocalHostName() + ":" +
+        common::NumToString(getpid()) + ":" +
+        common::NumToString(common::timer::now_time());
     return uuid;
 }
 
diff --git a/src/sdk/fs_impl.h b/src/sdk/fs_impl.h
index 9eec0714..b97006a8 100644
--- a/src/sdk/fs_impl.h
+++ b/src/sdk/fs_impl.h
@@ -32,8 +32,8 @@ class FSImpl : public FS {
     int32_t CreateDirectory(const char* path);
     int32_t ListDirectory(const char* path, BfsFileInfo** filelist, int *num);
     int32_t DeleteDirectory(const char* path, bool recursive);
-    int32_t LockDirectory(const char* path);
-    int32_t UnlockDirectory(const char* path);
+    int32_t LockDirectory(const char* path, int32_t timeout);
+    int32_t UnlockDirectory(const char* path, bool force_unloak = false);
     int32_t DiskUsage(const char* path, int64_t* du_size);
     int32_t Access(const char* path, int32_t mode);
     int32_t Stat(const char* path, BfsFileInfo* fileinfo);