func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	key := args.Key

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

	wantLease := args.WantLease
	hostPort := args.HostPort

	being := beingRevoked(ss, key)
	if wantLease && !being {
		print(ss.beingRevoked)
		makeLease(ss, key, hostPort)
		reply.Lease = storagerpc.Lease{Granted: true, ValidSeconds: storagerpc.LeaseSeconds}
	}

	ss.valMapLock.Lock()
	defer ss.valMapLock.Unlock()

	value, in := ss.valMap[key]
	if !in {
		reply.Status = storagerpc.KeyNotFound
	} else {
		reply.Status = storagerpc.OK
		reply.Value = value
	}

	return nil
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {

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

	if data, found := ss.topMap[key]; found {
		if str, ok := data.(string); ok {

			// key was found and had valid string data
			reply.Status = storagerpc.OK
			reply.Value = str
			return nil
		} else {

			// key value is corrupted, not a string
			return errors.New("bad value")
		}
	} else {
		reply.Status = storagerpc.KeyNotFound
		return nil
	}
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	if DBG {
		fmt.Println("Entered storage server get")
	}

	if !ss.inRange(libstore.StoreHash(args.Key)) {
		reply.Status = storagerpc.WrongServer
		return nil
	}
	ss.dataLock.Lock()
	val, ok := ss.dataStore[args.Key]
	ss.dataLock.Unlock()

	if !ok {
		reply.Status = storagerpc.KeyNotFound
	} else {
		reply.Status = storagerpc.OK
		reply.Value = val.(string)
		if args.WantLease && ss.pendingMap[args.Key].pending == 0 {
			ss.leaseLock.Lock()
			lease := storagerpc.Lease{Granted: true, ValidSeconds: storagerpc.LeaseSeconds}
			reply.Lease = lease
			// Track that this lease was issued
			leaseWrap := LeaseWrapper{lease: lease, timeGranted: time.Now(), hostport: args.HostPort}
			_, ok := ss.leaseStore[args.Key]
			if !ok {
				ss.leaseStore[args.Key] = list.New()
			}
			ss.leaseStore[args.Key].PushBack(leaseWrap)
			ss.leaseLock.Unlock()
		}
	}
	return nil
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) 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()
		value, exist := ss.userValue[args.Key]
		if exist == false {
			reply.Status = storagerpc.KeyNotFound
		} else {
			reply.Status = storagerpc.OK
			reply.Value = value
			if args.WantLease == true {
				ss.grantLease(args.HostPort, ss.userValueLease[args.Key], &reply.Lease)
			}
		}
	}
	return nil
}

func (ss *storageServer) addLeaseRecord(args *storagerpc.GetArgs, reply *storagerpc.GetReply) *storagerpc.GetReply {
	ss.rwl.Lock()
	defer ss.rwl.Unlock()

	// to refuse the lease request
	if ss.inRevoking[args.Key] {
		reply.Lease = storagerpc.Lease{Granted: false}
		return reply
	}

	// append a lease record
	leaseList := ss.leases[args.Key]
	if leaseList == nil {
		ss.leases[args.Key] = list.New()
		leaseList = ss.leases[args.Key]
	}

	leaseList.PushBack(&leaseRecord{
		expirationTime: time.Now().Add(time.Second * time.Duration(leaseSeconds)),
		hostport:       args.HostPort,
	})

	reply.Lease = storagerpc.Lease{
		Granted:      true,
		ValidSeconds: storagerpc.LeaseSeconds,
	}
	return reply
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	generalReply, value := ss.generalGet(args)

	reply.Status = generalReply.Status
	if reply.Status != storagerpc.OK {
		return nil
	}

	reply.Lease = generalReply.Lease
	reply.Value = value.(string)

	return nil
}

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

	// Step 1: Lock the keyLockMap access so as to find the lock for this specific key.
	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()

	// Step 2: Release the sMutex lock so that the ss can serve other GET requests.
	// Meanwhile, since we are dealing with lease related to args.Key, we must lock
	// it using its own lock, keyLock.
	granted := false
	keyLock.Lock()
	defer keyLock.Unlock()

	if args.WantLease {
		leasedLibStores, ok := ss.leaseMap[args.Key]
		if !ok {
			leasedLibStores = make(map[string]time.Time)
			ss.leaseMap[args.Key] = leasedLibStores
		}
		ss.leaseMap[args.Key][args.HostPort] = time.Now()
		granted = true

	}
	reply.Lease = storagerpc.Lease{granted, storagerpc.LeaseSeconds}

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

	if ok {
		reply.Value = val.value
		reply.Status = storagerpc.OK
	} else {
		reply.Status = storagerpc.KeyNotFound
	}
	return nil
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	if !ss.isInServerRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	//get the lock for current key
	ss.mutex.Lock()
	lock, exist := ss.lockMap[args.Key]
	if !exist {
		lock = new(sync.Mutex)
		ss.lockMap[args.Key] = lock
	}
	ss.mutex.Unlock()

	//Lock current key we are going to work on
	lock.Lock()

	//process lease request
	grantLease := false
	if args.WantLease {
		//add to lease map
		var libServerList *list.List
		libServerList, exist := ss.leaseMap[args.Key]
		if !exist {
			libServerList = new(list.List)
		}
		leaseRecord := LeaseRecord{args.HostPort, time.Now()}
		libServerList.PushBack(leaseRecord)
		ss.leaseMap[args.Key] = libServerList
		grantLease = true
	}
	reply.Lease = storagerpc.Lease{grantLease, storagerpc.LeaseSeconds}

	//retrieve value
	value, exist := ss.keyValueMap[args.Key]
	if !exist {
		reply.Status = storagerpc.KeyNotFound
	} else {
		reply.Status = storagerpc.OK
		reply.Value = value
	}

	lock.Unlock()
	return nil
}

func (ss *storageServer) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	if !ss.IsKeyInRange(args.Key) {
		reply.Status = storagerpc.WrongServer
		return nil
	}

	value, exist := ss.stringTable[args.Key]
	if !exist {
		reply.Status = storagerpc.KeyNotFound
	} else {
		reply.Status = storagerpc.OK
		reply.Value = value
		if args.WantLease {
			granted := ss.GrantLease(args.Key, args.HostPort)
			reply.Lease = storagerpc.Lease{granted, storagerpc.LeaseSeconds}
		}
	}
	//fmt.Println("Get Value ", value, " for key ", args.Key)
	return nil
}

func (pc *proxyCounter) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {
	if pc.override {
		reply.Status = pc.overrideStatus
		return pc.overrideErr
	}
	byteCount := len(args.Key)
	if args.WantLease {
		atomic.AddUint32(&pc.leaseRequestCount, 1)
	}
	if pc.disableLease {
		args.WantLease = false
	}
	err := pc.srv.Call("StorageServer.Get", args, reply)
	byteCount += len(reply.Value)
	if reply.Lease.Granted {
		if pc.overrideLeaseSeconds > 0 {
			reply.Lease.ValidSeconds = pc.overrideLeaseSeconds
		}
		atomic.AddUint32(&pc.leaseGrantedCount, 1)
	}
	atomic.AddUint32(&pc.rpcCount, 1)
	atomic.AddUint32(&pc.byteCount, uint32(byteCount))
	return err
}

// Get retrieves the specified key from the data store and replies with
// the key's value and a lease if one was requested. 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) Get(args *storagerpc.GetArgs, reply *storagerpc.GetReply) error {

	//	fmt.Println("Get function is called.", args);
	if FindPartitionNode(ss.ring, args.Key).NodeID != ss.nodeID {
		reply.Status = storagerpc.WrongServer
		fmt.Println("GetFunction: Wrong Server")
		return nil
	}

	// first get the value from the storage map
	ss.sMutex.Lock()
	data, ok := ss.storage[args.Key]
	ss.sMutex.Unlock()

	if !ok {
		// if the key is not found
		reply.Status = storagerpc.KeyNotFound
		//		fmt.Println("@@@@GetFunction: key not found!")
		return nil //errors.New("Key not found for Get Function.");
	} else {
		data.dMutex.Lock()
		defer data.dMutex.Unlock()

		if args.WantLease {
			//			fmt.Println("GetFunction: Granting Lease.",args.HostPort,args.Key);
			// want a lease
			// grant a lease
			reply.Status = storagerpc.OK
			reply.Value = data.data.(string)
			reply.Lease.Granted = true
			reply.Lease.ValidSeconds = storagerpc.LeaseSeconds

			if data.Set == nil {
				lMap := make(map[string](*leaseCallBack))
				data.Set = lMap
			}

			// add this one to the lease list
			var val *leaseCallBack
			val, ok = data.Set[args.HostPort]
			if ok {
				// only update the lease time, don't need to create the hanlder
				val.registerTime = time.Now().Unix()
			} else {
				// need to create a new leaseCallBack struct
				//				cli, err := rpc.DialHTTP("tcp", args.HostPort)
				//				if err != nil {
				//					fmt.Println("GetFunction add callback lease failed.")
				//					return err
				//				}

				val = &leaseCallBack{
					//					cli:          cli,
					cbAddr:       args.HostPort,
					registerTime: time.Now().Unix(),
					validation:   (storagerpc.LeaseGuardSeconds + storagerpc.LeaseSeconds)}

				data.Set[args.HostPort] = val
			}
		} else {
			reply.Status = storagerpc.OK
			reply.Value = data.data.(string)
			reply.Lease.Granted = false
		}
	}
	//	fmt.Println("Get function returns.", reply.Value);
	return nil
}