Aether Scaling

drsm/api.go

@@ -49,14 +49,19 @@ type DrsmInterface interface {
 	DeletePod(string)
 }
 
+// func InitDRSM(sharedPoolName string, myid PodId, db DbInfo, opt *Options, punchLivenessTime int) (DrsmInterface, error) {
 func InitDRSM(sharedPoolName string, myid PodId, db DbInfo, opt *Options) (DrsmInterface, error) {
 	logger.DrsmLog.Debugln("client id:", myid)
 
 	d := &Drsm{sharedPoolName: sharedPoolName,
 		clientId: myid,
 		db:       db,
 		mode:     ResourceClient}
-
+	// if punchLivenessTime == 0 {
+	d.punchLivenessTime = 5000
+	// } else {
+	// 	d.punchLivenessTime = punchLivenessTime
+	// }
 	d.ConstuctDrsm(opt)
 
 	return d, nil
@@ -110,8 +115,8 @@ func (d *Drsm) ReleaseInt32ID(id int32) error {
 }
 
 func (d *Drsm) FindOwnerInt32ID(id int32) (*PodId, error) {
-	d.globalChunkTblMutex.Lock()
-	defer d.globalChunkTblMutex.Unlock()
+	d.globalChunkTblMutex.RLock()
+	defer d.globalChunkTblMutex.RUnlock()
 	chunkId := id >> 10
 	chunk, found := d.globalChunkTbl[chunkId]
 	if found {

drsm/chunk.go

@@ -30,9 +30,9 @@ func (d *Drsm) GetNewChunk() (*chunk, error) {
 	for {
 		for {
 			cn = rand.Int31n(d.chunkIdRange)
-			d.globalChunkTblMutex.Lock()
+			d.globalChunkTblMutex.RLock()
 			_, found := d.globalChunkTbl[cn]
-			d.globalChunkTblMutex.Unlock()
+			d.globalChunkTblMutex.RUnlock()
 			if found {
 				continue
 			}

drsm/claim.go

@@ -17,9 +17,9 @@ func (d *Drsm) podDownDetected() {
 		// Given Pod find out current Chunks owned by this POD
 		pd := d.podMap[p]
 		for k := range pd.podChunks {
-			d.globalChunkTblMutex.Lock()
+			d.globalChunkTblMutex.RLock()
 			c, found := d.globalChunkTbl[k]
-			d.globalChunkTblMutex.Unlock()
+			d.globalChunkTblMutex.RUnlock()
 			logger.DrsmLog.Debugf("found: %v chunk: %v", found, c)
 			if found {
 				go c.claimChunk(d)

drsm/drsm.go

@@ -57,7 +57,8 @@ type Drsm struct {
 	ipModule            ipam.Ipamer
 	prefix              map[string]*ipam.Prefix
 	mongo               *MongoDBLibrary.MongoClient
-	globalChunkTblMutex sync.Mutex
+	globalChunkTblMutex sync.RWMutex
+	punchLivenessTime   int
 }
 
 func (d *Drsm) DeletePod(podInstance string) {
@@ -84,13 +85,17 @@ func (d *Drsm) ConstuctDrsm(opt *Options) {
 	d.podMap = make(map[string]*podData)
 	d.podDown = make(chan string, 10)
 	d.scanChunks = make(map[int32]*chunk)
-	d.globalChunkTblMutex = sync.Mutex{}
+	d.globalChunkTblMutex = sync.RWMutex{}
 	d.initIpam(opt)
 
 	//connect to DB
 	d.mongo, _ = MongoDBLibrary.NewMongoClient(d.db.Url, d.db.Name)
 	logger.DrsmLog.Debugln("mongoClient is created", d.db.Name)
+	_, err := d.mongo.CreateIndex(d.sharedPoolName, "_id")
 
+	if err != nil {
+		logger.AppLog.Infof("Failed to create id index: %v", err)
+	}
 	go d.handleDbUpdates()
 	go d.punchLiveness()
 	go d.podDownDetected()

drsm/updates.go

@@ -156,9 +156,9 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan
 				// looks like chunk owner getting change
 				owner := s.Update.UpdFields.PodId
 				c := getChunIdFromDocId(s.DId.Id)
-				d.globalChunkTblMutex.Lock()
+				d.globalChunkTblMutex.RLock()
 				cp := d.globalChunkTbl[c]
-				d.globalChunkTblMutex.Unlock()
+				d.globalChunkTblMutex.RUnlock()
 				// TODO update IP address as well.
 				cp.Owner.PodName = owner
 				cp.Owner.PodIp = s.Update.UpdFields.PodIp
@@ -184,6 +184,8 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan
 
 func (d *Drsm) punchLiveness() {
 	// write to DB - signature every 5 second
+	// or make it parameterized
+	// ticker := time.NewTicker(time.Duration(d.punchLivenessTime) * time.Millisecond)
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
 
@@ -199,6 +201,7 @@ func (d *Drsm) punchLiveness() {
 		//logger.DrsmLog.Debugln("update keepalive time")
 		filter := bson.M{"_id": d.clientId.PodName}
 
+		// timein := time.Now().Local().Add(time.Millisecond * time.Duration(d.punchLivenessTime+500))
 		timein := time.Now().Local().Add(20 * time.Second)
 
 		update := bson.D{