diff --git a/loader/loader.go b/loader/loader.go
index 48b0776a7733d546ab2b62cb0c36d8d808814c26..93ffc426499c802a8ef5d07ae62bed43fdf84e11 100644
--- a/loader/loader.go
+++ b/loader/loader.go
@@ -129,7 +129,7 @@ func (s *state) parseStream(done chan error) {
func (s *state) handleNQuads(wg *sync.WaitGroup) {
for nq := range s.cnq {
- edge, err := nq.ToEdge(s.instanceIdx, s.numInstances, uidStore)
+ edge, err := nq.ToEdge(s.instanceIdx, s.numInstances)
for err != nil {
// Just put in a retry loop to tackle temporary errors.
if err == posting.E_TMP_ERROR {
@@ -140,7 +140,7 @@ func (s *state) handleNQuads(wg *sync.WaitGroup) {
Error("While converting to edge")
return
}
- edge, err = nq.ToEdge(s.instanceIdx, s.numInstances, uidStore)
+ edge, err = nq.ToEdge(s.instanceIdx, s.numInstances)
}
// Only handle this edge if the attribute satisfies the modulo rule
@@ -159,7 +159,7 @@ func (s *state) handleNQuads(wg *sync.WaitGroup) {
}
func (s *state) getUidForString(str string) {
- _, err := rdf.GetUid(str, s.instanceIdx, s.numInstances, dataStore)
+ _, err := rdf.GetUid(str, s.instanceIdx, s.numInstances)
for err != nil {
// Just put in a retry loop to tackle temporary errors.
if err == posting.E_TMP_ERROR {
@@ -171,7 +171,7 @@ func (s *state) getUidForString(str string) {
Error("While getting UID")
return
}
- _, err = rdf.GetUid(str, s.instanceIdx, s.numInstances, dataStore)
+ _, err = rdf.GetUid(str, s.instanceIdx, s.numInstances)
}
}
@@ -197,7 +197,9 @@ func (s *state) handleNQuadsWhileAssign(wg *sync.WaitGroup) {
}
// Blocking function.
-func HandleRdfReader(reader io.Reader, instanceIdx uint64, numInstances uint64) (uint64, error) {
+func HandleRdfReader(reader io.Reader, instanceIdx uint64,
+ numInstances uint64) (uint64, error) {
+
s := new(state)
s.ctr = new(counters)
ticker := time.NewTicker(time.Second)
diff --git a/rdf/parse.go b/rdf/parse.go
index 8d59bb3447e294548ff84f5de050961cd7d81651..789553d47b3e1b6fa6e9f1d0befe6a3017efea42 100644
--- a/rdf/parse.go
+++ b/rdf/parse.go
@@ -23,7 +23,6 @@ import (
"time"
"github.com/dgraph-io/dgraph/lex"
- "github.com/dgraph-io/dgraph/store"
"github.com/dgraph-io/dgraph/uid"
"github.com/dgraph-io/dgraph/x"
)
@@ -37,23 +36,23 @@ type NQuad struct {
Language string
}
-func GetUid(s string, instanceIdx uint64, numInstances uint64,
- rStore *store.Store) (uint64, error) {
+func GetUid(s string, instanceIdx uint64, numInstances uint64) (uint64, error) {
if strings.HasPrefix(s, "_uid_:") {
return strconv.ParseUint(s[6:], 0, 64)
}
return uid.GetOrAssign(s, instanceIdx, numInstances)
}
-func (nq NQuad) ToEdge(instanceIdx, numInstances uint64,
- rStore *store.Store) (result x.DirectedEdge, rerr error) {
- sid, err := GetUid(nq.Subject, instanceIdx, numInstances, rStore)
+func (nq NQuad) ToEdge(instanceIdx,
+ numInstances uint64) (result x.DirectedEdge, rerr error) {
+
+ sid, err := GetUid(nq.Subject, instanceIdx, numInstances)
if err != nil {
return result, err
}
result.Entity = sid
if len(nq.ObjectId) > 0 {
- oid, err := GetUid(nq.ObjectId, instanceIdx, numInstances, rStore)
+ oid, err := GetUid(nq.ObjectId, instanceIdx, numInstances)
if err != nil {
return result, err
}
diff --git a/server/loader/main.go b/server/loader/main.go
index 8f6bcf7cc4ae3a4e5b54754c92f39a3a267b5a5b..3b5bc31fbb1a041db42524eb6f8d51b447219a88 100644
--- a/server/loader/main.go
+++ b/server/loader/main.go
@@ -43,7 +43,8 @@ var numInstances = flag.Uint64("numInstances", 1,
var postingDir = flag.String("postings", "", "Directory to store posting lists")
var uidDir = flag.String("uidDir", "", "Directory to read UID posting lists")
var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
-var numcpu = flag.Int("numCpu", runtime.NumCPU(), "Number of cores to be used by the process")
+var numcpu = flag.Int("numCpu", runtime.NumCPU(),
+ "Number of cores to be used by the process")
func main() {
flag.Parse()
diff --git a/server/uidassigner/main.go b/server/uidassigner/main.go
index cdf0d29d73ad42ef82a3029e1d7b564f524c7d6b..a2984b6682497dce4fef512944807389cfaf1776 100644
--- a/server/uidassigner/main.go
+++ b/server/uidassigner/main.go
@@ -24,9 +24,11 @@ var instanceIdx = flag.Uint64("instanceIdx", 0,
"Only pick entities, where Fingerprint % numInstance == instanceIdx.")
var numInstances = flag.Uint64("numInstances", 1,
"Total number of instances among which uid assigning is shared")
-var uidDir = flag.String("uidpostings", "", "Directory to store xid to uid posting lists")
+var uidDir = flag.String("uidpostings", "",
+ "Directory to store xid to uid posting lists")
var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
-var numcpu = flag.Int("numCpu", runtime.NumCPU(), "Number of cores to be used by the process")
+var numcpu = flag.Int("numCpu", runtime.NumCPU(),
+ "Number of cores to be used by the process")
func main() {
flag.Parse()
@@ -51,7 +53,7 @@ func main() {
glog.WithField("instanceIdx", *instanceIdx).
WithField("numInstances", *numInstances).
- Info("Only those XIDs with FP(xid) % numInstance == instanceIdx will be given UID")
+ Info("Only XIDs with FP(xid)%numInstance == instanceIdx will be given UID")
if len(*rdfGzips) == 0 {
glog.Fatal("No RDF GZIP files specified")
@@ -81,7 +83,8 @@ func main() {
glog.WithError(err).Fatal("Unable to create gzip reader.")
}
- count, err := loader.HandleRdfReaderWhileAssign(r, *instanceIdx, *numInstances)
+ count, err := loader.HandleRdfReaderWhileAssign(r, *instanceIdx,
+ *numInstances)
if err != nil {
glog.WithError(err).Fatal("While handling rdf reader.")
}
diff --git a/server/uidassigner/main_test.go b/server/uidassigner/main_test.go
index 269b9a839845955eb9114b67f08f1a2bcf31fd7f..85ec4aa69831725a9512dc70e85848edaf96b5e7 100644
--- a/server/uidassigner/main_test.go
+++ b/server/uidassigner/main_test.go
@@ -41,14 +41,14 @@ func TestQuery(t *testing.T) {
list := []string{"alice", "bob", "mallory", "ash", "man", "dgraph"}
for _, str := range list {
if farm.Fingerprint64([]byte(str))%numInstances == 0 {
- uid, err := rdf.GetUid(str, 0, numInstances, ps)
+ uid, err := rdf.GetUid(str, 0, numInstances)
if uid < minIdx0 || uid > minIdx0+mod-1 {
t.Error("Not the correct UID", err)
}
t.Logf("Instance-0 Correct UID", str, uid)
} else {
- uid, err := rdf.GetUid(str, 1, numInstances, ps)
+ uid, err := rdf.GetUid(str, 1, numInstances)
if uid < minIdx1 || uid > minIdx1+mod-1 {
t.Error("Not the correct UID", err)
}
diff --git a/uid/assigner.go b/uid/assigner.go
index fad1d5740423da27e6040d1088f0c27ad34eb1e9..ef835558cfb37e7e61065d44821283acc11a8999 100644
--- a/uid/assigner.go
+++ b/uid/assigner.go
@@ -98,11 +98,11 @@ func Init(ps *store.Store) {
uidStore = ps
}
-func allocateUniqueUid(xid string, instanceIdx uint64, numInstances uint64) (uid uint64, rerr error) {
+func allocateUniqueUid(xid string, instanceIdx uint64,
+ numInstances uint64) (uid uint64, rerr error) {
mod := math.MaxUint64 / numInstances
minIdx := instanceIdx * mod
-
for sp := ""; ; sp += " " {
txid := xid + sp
@@ -148,6 +148,7 @@ func allocateUniqueUid(xid string, instanceIdx uint64, numInstances uint64) (uid
func assignNew(pl *posting.List, xid string, instanceIdx uint64,
numInstances uint64) (uint64, error) {
+
entry := lmgr.newOrExisting(xid)
entry.Lock()
entry.ts = time.Now()
@@ -187,7 +188,9 @@ func stringKey(xid string) []byte {
return buf.Bytes()
}
-func GetOrAssign(xid string, instanceIdx uint64, numInstances uint64) (uid uint64, rerr error) {
+func GetOrAssign(xid string, instanceIdx uint64,
+ numInstances uint64) (uid uint64, rerr error) {
+
key := stringKey(xid)
pl := posting.GetOrCreate(key, uidStore)
if pl.Length() == 0 {