<commit_before>// This script is used to load data into Dgraph from an RDF file by performing
// mutations using the HTTP interface.
//
// You can run the script like
// go build . && ./mutations --rdf $GOPATH/src/github.com/dgraph-io/benchmarks/data/names.gz --concurrent 100
package main

import (
	"bufio"
	"compress/gzip"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/dgraph-io/dgraph/x"
)

var (
	file       = flag.String("rdf", "", "Location of rdf file to load")
	dgraph     = flag.String("dgraph", "http://127.0.0.1:8080/query", "Dgraph server address")
	concurrent = flag.Int("concurrent", 1000, "Number of concurrent requests to make to Dgraph")
	numRdf     = flag.Int("mutations", 100, "Number of RDF N-Quads to send as part of a mutation.")
)

func body(rdf string) string {
	return `mutation {
  set {
  	` + rdf + `
      }
  }`
}

type response struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

var hc http.Client
var r response

func makeRequest(mutation chan string, c *uint64, wg *sync.WaitGroup) {
	var counter uint64
	for m := range mutation {
		counter = atomic.AddUint64(c, 1)
		if counter%10000 == 0 {
			fmt.Printf("Request: %v\n", counter)
		}
		req, err := http.NewRequest("POST", *dgraph, strings.NewReader(body(m)))
		x.Check(err)
		res, err := hc.Do(req)
		x.Check(err)

		body, err := ioutil.ReadAll(res.Body)
		x.Check(err)
		x.Check(json.Unmarshal(body, &r))
		if r.Code != "ErrorOk" {
			log.Fatalf("Error while performing mutation: %v, err: %v", m, r.Message)
		}
	}
	wg.Done()
}

func buildMutation(rdf chan string, m chan string, rdfCount *uint64, wg *sync.WaitGroup) {
	set := ""
	count := 0
	for r := range rdf {
		atomic.AddUint64(rdfCount, 1)
		count++
		set += r + "\n"
		if count == *numRdf {
			m <- set
			set = ""
			count = 0
		}

	}
	if set != "" {
		m <- set
	}
	wg.Done()
}

func main() {
	flag.Parse()
	f, err := os.Open(*file)
	x.Check(err)
	defer f.Close()
	gr, err := gzip.NewReader(f)
	x.Check(err)

	hc = http.Client{Timeout: 5 * time.Minute}

	rdf := make(chan string, 1000)
	mutation := make(chan string, 100)

	var rdfCount uint64 = 0
	var wgm sync.WaitGroup
	for i := 0; i < 10; i++ {
		wgm.Add(1)
		go buildMutation(rdf, mutation, &rdfCount, &wgm)
	}

	var count uint64 = 0
	var wg sync.WaitGroup
	for i := 0; i < *concurrent; i++ {
		wg.Add(1)
		go makeRequest(mutation, &count, &wg)
	}

	scanner := bufio.NewScanner(gr)
	for scanner.Scan() {
		// Lets send the rdf's to the channel.
		rdf <- scanner.Text()
	}
	x.Check(scanner.Err())
	close(rdf)

	wgm.Wait()
	close(mutation)
	wg.Wait()
	fmt.Println("Number of RDF's parsed: ", rdfCount)
	fmt.Println("Number of mutations run: ", count)
}<commit_msg>Simplify code and shorten flags.<commit_after>// This script is used to load data into Dgraph from an RDF file by performing
// mutations using the HTTP interface.
//
// You can run the script like
// go build . && ./mutations --rdf $GOPATH/src/github.com/dgraph-io/benchmarks/data/names.gz --concurrent 100
package main

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/dgraph-io/dgraph/x"
)

var (
	file       = flag.String("r", "", "Location of rdf file to load")
	dgraph     = flag.String("d", "http://127.0.0.1:8080/query", "Dgraph server address")
	concurrent = flag.Int("c", 1000, "Number of concurrent requests to make to Dgraph")
	numRdf     = flag.Int("m", 100, "Number of RDF N-Quads to send as part of a mutation.")
)

func body(rdf string) string {
	return fmt.Sprintf("mutation { set { %s } }", rdf)
}

type response struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

var hc http.Client
var r response

func makeRequest(mutation chan string, c *uint64, wg *sync.WaitGroup) {
	var counter uint64
	for m := range mutation {
		counter = atomic.AddUint64(c, 1)
		if counter%100 == 0 {
			fmt.Printf("Request: %v\n", counter)
		}
		req, err := http.NewRequest("POST", *dgraph, strings.NewReader(body(m)))
		x.Check(err)
		res, err := hc.Do(req)
		x.Check(err)

		body, err := ioutil.ReadAll(res.Body)
		x.Check(err)
		x.Check(json.Unmarshal(body, &r))
		if r.Code != "ErrorOk" {
			log.Fatalf("Error while performing mutation: %v, err: %v", m, r.Message)
		}
	}
	wg.Done()
}

func main() {
	flag.Parse()
	f, err := os.Open(*file)
	x.Check(err)
	defer f.Close()
	gr, err := gzip.NewReader(f)
	x.Check(err)

	hc = http.Client{Timeout: time.Minute}
	mutation := make(chan string, 3*(*concurrent))

	var count uint64 = 0
	var wg sync.WaitGroup
	for i := 0; i < *concurrent; i++ {
		wg.Add(1)
		go makeRequest(mutation, &count, &wg)
	}

	var buf bytes.Buffer
	scanner := bufio.NewScanner(gr)
	num := 0
	var rdfCount uint64 = 0
	for scanner.Scan() {
		buf.WriteString(scanner.Text())
		buf.WriteRune('\n')

		if num >= *numRdf {
			mutation <- buf.String()
			buf.Reset()
			num = 0
		}
		rdfCount++
		num++
	}
	x.Check(scanner.Err())
	if buf.Len() > 0 {
		mutation <- buf.String()
	}
	close(mutation)

	wg.Wait()
	fmt.Println("Number of RDF's parsed: ", rdfCount)
	fmt.Println("Number of mutations run: ", count)
}
{'fraction_non_alphanumeric': 0.16367112810707457, 'fraction_numerical': 0.019694072657743786, 'mean_word_length': 3.3374792703150913, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4797467', 'n_tokens_mistral': 2248, 'n_tokens_neox': 2047, 'n_words': 548}