func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {
	if !ss.IsKeyInRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.lockForTableLocks.Lock()
	/* Lock table for writing */
	lock, exist := ss.tableLocks[args.Key]
	if !exist {
		lock = new(sync.Mutex)
		ss.tableLocks[args.Key] = lock
	}
	ss.lockForTableLocks.Unlock()
	lock.Lock()
	defer lock.Unlock()

	ss.RevokeLease(args.Key)

	_, exist = ss.stringTable[args.Key]
	if !exist {
		reply.Status = storagerpc.KeyNotFound
	} else {

		delete(ss.stringTable, args.Key)
		reply.Status = storagerpc.OK
	}
	return nil
}

func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {

	key := args.Key

	if !correctServer(ss, key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	leases := getLeases(ss, key)
	if leases != nil {
		ss.editLock.Lock()
		revokeLease(ss, key)
		ss.editLock.Unlock()
	}
	ss.valMapLock.Lock()
	defer ss.valMapLock.Unlock()

	_, in := ss.valMap[key]
	if !in {
		reply.Status = storagerpc.KeyNotFound
	} else {
		delete(ss.valMap, key)
		reply.Status = storagerpc.OK
	}

	return nil

}

// Delete remove the specified key from the data store.
// If the key does not fall within the storage server's range,
// it should reply with status WrongServer.
// If the key is not found, it should reply with status KeyNotFound.
func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {

	//	fmt.Println("Delete function Called",args)
	if FindPartitionNode(ss.ring, args.Key).NodeID != ss.nodeID {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	// lock the whole storage first
	ss.sMutex.Lock()
	defer ss.sMutex.Unlock()

	// retrieve the data unit
	data, ok := ss.storage[args.Key]
	if ok {
		reply.Status = storagerpc.OK
		data.dMutex.Lock()
		// WARNING: here we might meet the situation of wild pointer.
		// that is the value is already deleted from the map, but the
		ss.RevokeCallBacksForKey(args.Key, data)
		data.dMutex.Unlock()

		// aplly changes
		delete(ss.storage, args.Key)

		return nil
	} else {
		reply.Status = storagerpc.KeyNotFound
		return nil
	}

	return nil
}

func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {
	if ss.isReady == false {
		reply.Status = storagerpc.NotReady
	} else {
		if ss.getServerID(args.Key) != ss.nodeID {
			reply.Status = storagerpc.WrongServer
			return nil
		}
		ss.serverMutex.Lock()
		if ss.userValueMutex[args.Key] == nil {
			ss.userValueMutex[args.Key] = &sync.Mutex{}
		}
		ss.serverMutex.Unlock()
		ss.userValueMutex[args.Key].Lock()
		defer ss.userValueMutex[args.Key].Unlock()
		_, exist := ss.userValue[args.Key]
		if exist == false {
			reply.Status = storagerpc.KeyNotFound
		} else {
			// revoke all leases if already exist in server before
			err := ss.revokeLease(args.Key, ss.userValueLease[args.Key])
			if err != nil {
				return err
			}
			ss.userValueLease[args.Key] = make(map[string]*lease)

			reply.Status = storagerpc.OK
			delete(ss.userValue, args.Key)
		}
	}
	return nil
}

func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {
	if DBG {
		fmt.Println("Entered storageserver delete")
	}
	if !ss.inRange(libstore.StoreHash(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	_, ok := ss.dataStore[args.Key]
	if !ok {
		reply.Status = storagerpc.KeyNotFound
		return nil
	}

	// Leasing check
	ss.leaseLock.Lock()
	pendingModifications, ok := ss.pendingMap[args.Key]

	if !ok {
		reply.Status = storagerpc.KeyNotFound
		ss.leaseLock.Unlock()
		return nil
	}

	pendingModifications.pending++
	leaseHolders, ok := ss.leaseStore[args.Key]
	ss.leaseLock.Unlock()

	if pendingModifications.pending > 1 { // Block until it's our turn to modify key
		response := make(chan int)
		pendingModifications.pendingCh <- response
		<-response
	}

	if ok {
		ss.revokeLeases(leaseHolders, args.Key)
	}

	ss.dataLock.Lock()
	delete(ss.dataStore, args.Key)
	ss.dataLock.Unlock()
	reply.Status = storagerpc.OK
	pendingModifications.pending--
Loop:
	for {
		select {
		case ch := <-pendingModifications.pendingCh:
			ch <- 1
			break Loop
		default:
			break Loop
		}
	}
	return nil
}

func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {
	if args == nil {
		return errors.New("ss: Can't delete nil K/V pair")
	}
	if reply == nil {
		return errors.New("ss: Can't reply with nil in Delete")
	}
	if !(ss.CheckKeyInRange(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	ss.sMutex.Lock()
	keyLock, exist := ss.keyLockMap[args.Key]
	if !exist {
		// Create new lock for the key
		keyLock = &sync.Mutex{}
		ss.keyLockMap[args.Key] = keyLock
	}
	ss.sMutex.Unlock()

	keyLock.Lock()
	defer keyLock.Unlock()

	_, ok := ss.sMap[args.Key]

	if !ok {
		reply.Status = storagerpc.KeyNotFound
		return nil
	}

	hpTimeMap, leaseExists := ss.leaseMap[args.Key]
	if leaseExists {
		// Revoke all issued leases.
		successChan := make(chan int, 1)
		finishChan := make(chan int, 1)
		expected := len(hpTimeMap)
		go ss.CheckRevokeStatus(args.Key, successChan, finishChan, expected)
		for hp, _ := range hpTimeMap {
			go ss.RevokeLeaseAt(hp, args.Key, successChan)
		}

		<-finishChan

		delete(ss.leaseMap, args.Key)
	}

	delete(ss.sMap, args.Key)
	reply.Status = storagerpc.OK
	return nil
}

func (pc *proxyCounter) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {
	if pc.override {
		reply.Status = pc.overrideStatus
		return pc.overrideErr
	}
	byteCount := len(args.Key)
	err := pc.srv.Call("StorageServer.Delete", args, reply)
	atomic.AddUint32(&pc.rpcCount, 1)
	atomic.AddUint32(&pc.byteCount, uint32(byteCount))
	return err
}

func (ss *storageServer) Delete(args *storagerpc.DeleteArgs, reply *storagerpc.DeleteReply) error {

	key := args.Key
	if rightStorageServer(ss, key) == false {
		reply.Status = storagerpc.WrongServer

		return nil
	}
	if _, found := ss.keyLocks[key]; found == false {
		ss.keyLocks[key] = make(chan int, 1)
	} else {
		<-ss.keyLocks[key]
	}

	if _, found := ss.topMap[key]; found {
		delete(ss.topMap, key)
		reply.Status = storagerpc.OK
	} else {
		reply.Status = storagerpc.KeyNotFound
	}
	ss.keyLocks[key] <- 1
	return nil
}