client.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861
  1. package main
  2. /*
  3. #cgo CFLAGS: -O2
  4. #cgo LDFLAGS: -lcrypto -lm
  5. #include "../c/dpf.h"
  6. #include "../c/okvClient.h"
  7. #include "../c/dpf.c"
  8. #include "../c/okvClient.c"
  9. */
  10. import "C"
  11. import (
  12. "2PPS/lib"
  13. "bufio"
  14. "bytes"
  15. "crypto/rand"
  16. "crypto/sha256"
  17. "crypto/tls"
  18. "encoding/json"
  19. "fmt"
  20. "log"
  21. "math/big"
  22. mr "math/rand"
  23. "net"
  24. "os"
  25. "sort"
  26. "strconv"
  27. "strings"
  28. "sync"
  29. "time"
  30. "unsafe"
  31. "golang.org/x/crypto/nacl/box"
  32. )
  33. type tweet struct {
  34. Topics []string
  35. Text string
  36. }
  37. const leader string = "127.0.0.1:4441"
  38. //needs to be changed at leader/client at the same time
  39. const numClients = 1000
  40. //mylimit=8000
  41. //sudo prlimit --nofile=$mylimit --pid $$; ulimit -n $mylimit
  42. //for every terminal
  43. const dataLength int = 256
  44. //Maximum Transport Unit
  45. const mtu int = 1100
  46. var dbWriteSize int
  47. var round int
  48. var topicList []string
  49. var archiveTopicList []string
  50. var neededSubscriptions int
  51. var publisherAmount int
  52. var goodPadding int
  53. var blocksReceived int
  54. var timeBounds []float64
  55. //this translates to a simulated round length of ~2h
  56. var speedUp float64 = 7200 / ((maxTimePerRound.Seconds()) * 3)
  57. var maxTimePerRound time.Duration = 1000 * time.Second
  58. var startTime int
  59. var archiveInterests = make([]int, 1)
  60. var sharedSecret [numClients][2][32]byte = createSharedSecret()
  61. var wantsArchive = make([]byte, 1)
  62. var leaderPublicKey *[32]byte
  63. var followerPublicKey *[32]byte
  64. var clientPrivateKey [numClients]*[32]byte
  65. var clientPublicKey [numClients]*[32]byte
  66. func main() {
  67. f, err := os.OpenFile("evalDataClient", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
  68. if err != nil {
  69. log.Fatalf("error opening file: %v", err)
  70. }
  71. defer f.Close()
  72. log.SetOutput(f)
  73. log.Println("simulated Duration", speedUp)
  74. wg := &sync.WaitGroup{}
  75. getTimeBounds()
  76. for i := 0; i < numClients; i++ {
  77. wg.Add(1)
  78. go client(i, f)
  79. time.Sleep(1 * time.Millisecond)
  80. }
  81. wg.Wait()
  82. }
  83. func client(clientNumber int, f *os.File) {
  84. generatedPublicKey, generatedPrivateKey, err := box.GenerateKey(rand.Reader)
  85. if err != nil {
  86. panic(err)
  87. }
  88. clientPrivateKey[clientNumber] = generatedPrivateKey
  89. clientPublicKey[clientNumber] = generatedPublicKey
  90. C.initializeCipher()
  91. //initializes the connection to the leader
  92. conf := &tls.Config{
  93. InsecureSkipVerify: true,
  94. }
  95. leaderConn, err := tls.Dial("tcp", leader, conf)
  96. if err != nil {
  97. panic(err)
  98. }
  99. leaderConn.SetDeadline(time.Time{})
  100. //receives topics first so client can participate asap
  101. receiveTopicLists(leaderConn)
  102. //gets the public keys of both servers
  103. var tmpLeaderPubKey [32]byte
  104. _, err = leaderConn.Read(tmpLeaderPubKey[:])
  105. if err != nil {
  106. panic(err)
  107. }
  108. leaderPublicKey = &tmpLeaderPubKey
  109. var tmpFollowerPubKey [32]byte
  110. _, err = leaderConn.Read(tmpFollowerPubKey[:])
  111. if err != nil {
  112. panic(err)
  113. }
  114. followerPublicKey = &tmpFollowerPubKey
  115. //sends own public key
  116. writeTo(leaderConn, clientPublicKey[clientNumber][:])
  117. neededSubscriptionsBytes := readFrom(leaderConn, 4)
  118. neededSubscriptions = byteToInt(neededSubscriptionsBytes)
  119. startTimeBytes := readFrom(leaderConn, 4)
  120. startTime = byteToInt(startTimeBytes)
  121. //setup ends above
  122. //while client is active he is always connected and has to participate
  123. for {
  124. //gets current phase
  125. phase := readFrom(leaderConn, 1)
  126. if phase[0] == 1 {
  127. //gets current dbWriteSize from leader
  128. dbWriteSizeBytes := readFrom(leaderConn, 4)
  129. dbWriteSize = byteToInt(dbWriteSizeBytes)
  130. //roundAsBytes := readFrom(leaderConn, 4)
  131. roundAsBytes := make([]byte, 4)
  132. _, err = leaderConn.Read(roundAsBytes)
  133. if err != nil {
  134. panic(err)
  135. }
  136. round = byteToInt(roundAsBytes)
  137. if clientNumber == 0 {
  138. fmt.Println("Round ", round)
  139. }
  140. //request virtualAddress from leader via pirQuery
  141. encryptedQueryLeader, encryptedQueryFollower := createAuditPIRQuery(clientNumber)
  142. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  143. pos := receiveVirtualAddress(sharedSecret[clientNumber], leaderConn)
  144. tweet := getRealTweet(clientNumber)
  145. if clientNumber == numClients-1 {
  146. log.Println("Round", round)
  147. log.Println("publisherAmount", publisherAmount)
  148. log.Println("goodPadding", goodPadding)
  149. log.Println("blocksReceived", blocksReceived)
  150. log.Println("goodPadding Percentage", float64(goodPadding)/float64(blocksReceived))
  151. publisherAmount = 0
  152. }
  153. //prep the query
  154. dataSize := len(tweet)
  155. querySize := make([]byte, 4)
  156. cQuerySize := C.int(byteToInt(querySize))
  157. var dpfQueryA *C.uchar
  158. var dpfQueryB *C.uchar
  159. C.prepQuery(C.int(pos), C.int(dbWriteSize), (*C.uchar)(&tweet[0]), C.int(dataSize), &cQuerySize, &dpfQueryA, &dpfQueryB)
  160. intQuerySize := int(cQuerySize) //byteToInt(querySize)
  161. //write the query
  162. queryAPlaintext := C.GoBytes(unsafe.Pointer(dpfQueryA), C.int(intQuerySize))
  163. //encrypts queryA and appends it to message
  164. var nonce [24]byte
  165. //fill nonce with randomness
  166. _, err = rand.Read(nonce[:])
  167. if err != nil {
  168. panic("couldn't get randomness for nonce!")
  169. }
  170. dpfQueryAEncrypted := box.Seal(nonce[:], queryAPlaintext, &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  171. //encrypts queryB and appends it to message
  172. queryBPlaintext := C.GoBytes(unsafe.Pointer(dpfQueryB), C.int(intQuerySize))
  173. //fill nonce with randomness
  174. _, err = rand.Read(nonce[:])
  175. if err != nil {
  176. panic("couldn't get randomness for nonce!")
  177. }
  178. dpfQueryBEncrypted := box.Seal(nonce[:], queryBPlaintext, &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  179. //writes the dpfQuery to the leader
  180. dpfLengthBytes := intToByte(len(dpfQueryAEncrypted))
  181. writeTo(leaderConn, dpfLengthBytes)
  182. writeTo(leaderConn, dpfQueryAEncrypted)
  183. writeTo(leaderConn, dpfQueryBEncrypted)
  184. C.free(unsafe.Pointer(dpfQueryA))
  185. C.free(unsafe.Pointer(dpfQueryB))
  186. } else if phase[0] == 3 {
  187. /*
  188. possible Values
  189. 0 : new client
  190. leader expects sharedSecrets, expects pirQuery
  191. 1 : update needed
  192. leader sends topicList, performs local update of sharedSecret, expects pirQuery
  193. 2 : no update needed
  194. nothing
  195. */
  196. subPhase := readFrom(leaderConn, 1)
  197. var encryptedQueryLeader, encryptedQueryFollower []byte
  198. //first time participating
  199. if subPhase[0] == 0 {
  200. receiveTopicLists(leaderConn)
  201. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(int(subPhase[0]), clientNumber)
  202. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  203. }
  204. //updates the topic list and what client is interested in
  205. if subPhase[0] == 1 {
  206. receiveTopicLists(leaderConn)
  207. //updates local secret
  208. for index := 0; index < 2; index++ {
  209. sharedSecret[clientNumber][index] = sha256.Sum256(sharedSecret[clientNumber][index][:])
  210. }
  211. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(int(subPhase[0]), clientNumber)
  212. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, false)
  213. }
  214. receiveTweets(sharedSecret[clientNumber], leaderConn, false, clientNumber)
  215. if len(archiveTopicList) > 0 {
  216. wantsArchive[0] = 0
  217. } else {
  218. wantsArchive[0] = 0
  219. }
  220. writeTo(leaderConn, wantsArchive)
  221. if wantsArchive[0] == 1 && len(archiveTopicList) > 0 {
  222. encryptedQueryLeader, encryptedQueryFollower = createPIRQuery(-1, clientNumber)
  223. sendQuerys(encryptedQueryLeader, encryptedQueryFollower, leaderConn, true)
  224. receiveTweets(sharedSecret[clientNumber], leaderConn, true, clientNumber)
  225. }
  226. } else {
  227. fmt.Println("Phase", phase)
  228. panic("somethin went wrong")
  229. }
  230. }
  231. }
  232. //creates and sends the pirQuerys for each server
  233. func createPIRQuery(subPhase int, clientNumber int) ([]byte, []byte) {
  234. //later this will be taken from gui, this is only for testing
  235. topicsOfInterest := make([]int, 1)
  236. topicsOfInterest[0] = mr.Intn(10)
  237. archiveInterests[0] = mr.Intn(10)
  238. tmpNeededSubscriptions := neededSubscriptions
  239. if tmpNeededSubscriptions > len(topicList) {
  240. tmpNeededSubscriptions = len(topicList)
  241. }
  242. tmptopicsOfInterest := make([]int, len(topicsOfInterest))
  243. copy(tmptopicsOfInterest, topicsOfInterest)
  244. tmpTopicList := make([]string, len(topicList))
  245. copy(tmpTopicList, topicList)
  246. if wantsArchive[0] == 1 && subPhase == -1 {
  247. tmpNeededSubscriptions = len(archiveInterests)
  248. if tmpNeededSubscriptions > len(archiveTopicList) {
  249. tmpNeededSubscriptions = len(archiveTopicList)
  250. }
  251. tmptopicsOfInterest = archiveInterests //!todo take archiveInterests from gui
  252. tmpTopicList = archiveTopicList
  253. }
  254. //creates fake topicsOfInterest if client is boooring
  255. if len(tmptopicsOfInterest) < tmpNeededSubscriptions && subPhase != -1 {
  256. tmptopicsOfInterest = addFakeInterests(len(tmpTopicList), tmptopicsOfInterest, false)
  257. }
  258. //pirQuery [topicsOfInterest][serverAmount][topicAmount]byte
  259. pirQuerys := make([][][]byte, len(tmptopicsOfInterest))
  260. for i := range pirQuerys {
  261. pirQuerys[i] = make([][]byte, 2)
  262. for j := range pirQuerys[i] {
  263. pirQuerys[i][j] = make([]byte, len(tmpTopicList))
  264. }
  265. }
  266. //for leader
  267. //pirQuery will be filled with random bits
  268. for topic := range tmptopicsOfInterest {
  269. for index := range tmpTopicList {
  270. bit, err := rand.Int(rand.Reader, big.NewInt(2))
  271. if err != nil {
  272. panic(err)
  273. }
  274. pirQuerys[topic][0][index] = byte(bit.Int64())
  275. }
  276. }
  277. tmptopicsOfInterestBytes := make([]byte, len(tmpTopicList))
  278. for index := range tmptopicsOfInterest {
  279. if tmptopicsOfInterest[index] == 1 {
  280. tmptopicsOfInterestBytes[index] = 1
  281. }
  282. }
  283. for topicIndex, topic := range tmptopicsOfInterest {
  284. for index := range tmpTopicList {
  285. if topic == index {
  286. if pirQuerys[topicIndex][0][index] == 1 {
  287. pirQuerys[topicIndex][1][index] = 0
  288. } else {
  289. pirQuerys[topicIndex][1][index] = 1
  290. }
  291. } else {
  292. if pirQuerys[topicIndex][0][index] == 0 {
  293. pirQuerys[topicIndex][1][index] = 0
  294. } else {
  295. pirQuerys[topicIndex][1][index] = 1
  296. }
  297. }
  298. }
  299. }
  300. //flattens the querys to be able to send them more efficently
  301. messagesFlattened := make([][]byte, 2)
  302. //adds the sharedSecret to the first pirQuery when first time participating
  303. if subPhase == 0 {
  304. for server := 0; server < 2; server++ {
  305. messagesFlattened[server] = append(messagesFlattened[server], sharedSecret[clientNumber][server][:]...)
  306. }
  307. }
  308. for server := range messagesFlattened {
  309. for topic := range pirQuerys {
  310. messagesFlattened[server] = append(messagesFlattened[server], pirQuerys[topic][server]...)
  311. }
  312. }
  313. var nonce [24]byte
  314. _, err := rand.Read(nonce[:])
  315. if err != nil {
  316. panic("couldn't get randomness for nonce!")
  317. }
  318. encryptedQueryLeader := box.Seal(nonce[:], messagesFlattened[0], &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  319. _, err = rand.Read(nonce[:])
  320. if err != nil {
  321. panic("couldn't get randomness for nonce!")
  322. }
  323. encryptedQueryFollower := box.Seal(nonce[:], messagesFlattened[1], &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  324. return encryptedQueryLeader, encryptedQueryFollower
  325. }
  326. func sendQuerys(encryptedQueryLeader, encryptedQueryFollower []byte, leaderConn net.Conn, getArchive bool) {
  327. encryptedLength := len(encryptedQueryLeader)
  328. //sends the pirQuerysLength to the leader
  329. writeTo(leaderConn, intToByte(encryptedLength))
  330. //sends the pirQuerys to the leader
  331. writeTo(leaderConn, encryptedQueryLeader)
  332. writeTo(leaderConn, encryptedQueryFollower)
  333. if getArchive {
  334. writeTo(leaderConn, intToByte(len(archiveInterests)))
  335. }
  336. }
  337. func receiveVirtualAddress(sharedSecret [2][32]byte, leaderConn net.Conn) int {
  338. virtualAddressByte := readFrom(leaderConn, 4)
  339. //xores the sharedSecret
  340. for h := 0; h < 2; h++ {
  341. for i := 0; i < 4; i++ {
  342. virtualAddressByte[i] = virtualAddressByte[i] ^ sharedSecret[h][i]
  343. }
  344. }
  345. return byteToInt(virtualAddressByte)
  346. }
  347. func receiveTweets(sharedSecret [2][32]byte, leaderConn net.Conn, getArchive bool, clientNumber int) {
  348. tmpNeededSubscriptions := neededSubscriptions
  349. if tmpNeededSubscriptions > len(topicList) {
  350. tmpNeededSubscriptions = len(topicList)
  351. }
  352. if getArchive {
  353. tmpNeededSubscriptions = len(archiveInterests)
  354. if tmpNeededSubscriptions > len(archiveTopicList) {
  355. tmpNeededSubscriptions = len(archiveTopicList)
  356. }
  357. }
  358. for i := 0; i < tmpNeededSubscriptions; i++ {
  359. if !getArchive {
  360. blocksReceived++
  361. }
  362. //client receives tweets
  363. tweetsLengthBytes := readFrom(leaderConn, 4)
  364. tweetsLength := byteToInt(tweetsLengthBytes)
  365. tweets := readFrom(leaderConn, tweetsLength)
  366. //expand sharedSecret so it is of right length
  367. expandBy := len(tweets) / 32
  368. expandedSharedSecrets := make([][]byte, 2)
  369. for i := 0; i < 2; i++ {
  370. for j := 0; j < expandBy; j++ {
  371. expandedSharedSecrets[i] = append(expandedSharedSecrets[i], sharedSecret[i][:]...)
  372. }
  373. }
  374. //xors the received messge into the message to display
  375. for i := 0; i < 2; i++ {
  376. lib.Xor(expandedSharedSecrets[i][:], tweets)
  377. }
  378. index := strings.Index(string(tweets), ";;")
  379. if index != -1 {
  380. textArr := strings.Split(string(tweets), ";;;")
  381. text := textArr[:len(textArr)-1]
  382. if text[1] != "" {
  383. text[1] = text[1][1:]
  384. }
  385. ok := strings.Contains(text[0], text[1])
  386. if !ok {
  387. goodPadding++
  388. }
  389. } else if index == -1 && tweets[0] != 0 {
  390. fmt.Println("error")
  391. fmt.Println("round", round, string(tweets), "length", len(tweets))
  392. return
  393. //panic("received text not of correct format")
  394. }
  395. }
  396. }
  397. //creates a shared secret for each server
  398. func createSharedSecret() [numClients][2][32]byte {
  399. var tmpSharedSecret [numClients][2][32]byte
  400. for i := 0; i < numClients; i++ {
  401. for j := 0; j < 2; j++ {
  402. _, err := rand.Read(tmpSharedSecret[i][j][:])
  403. if err != nil {
  404. panic("couldn't get randomness for sharedSecret!")
  405. }
  406. }
  407. }
  408. return tmpSharedSecret
  409. }
  410. func createAuditPIRQuery(clientNumber int) ([]byte, []byte) {
  411. //pirQuery [serverAmount][dbWriteSize]byte
  412. pirQuerys := make([][]byte, 2)
  413. for i := range pirQuerys {
  414. pirQuerys[i] = make([]byte, dbWriteSize)
  415. }
  416. //for leader
  417. //pirQuery will be filled with random bits
  418. for index := range pirQuerys[0] {
  419. bit := mr.Intn(2)
  420. pirQuerys[0][index] = byte(bit)
  421. }
  422. copy(pirQuerys[1], pirQuerys[0])
  423. //the positon the virtual address will be taken from
  424. pos := mr.Intn(dbWriteSize)
  425. pirQuerys[0][pos] = 1
  426. pirQuerys[1][pos] = 0
  427. //flattens the querys to be able to send them more efficently
  428. messagesFlattened := make([][]byte, 2)
  429. //adds the sharedSecret to the pirQuery
  430. for server := 0; server < 2; server++ {
  431. messagesFlattened[server] = append(messagesFlattened[server], sharedSecret[clientNumber][server][:]...)
  432. }
  433. for server := 0; server < 2; server++ {
  434. messagesFlattened[server] = append(messagesFlattened[server], pirQuerys[server][:]...)
  435. }
  436. var nonce [24]byte
  437. _, err := rand.Read(nonce[:])
  438. if err != nil {
  439. panic("couldn't get randomness for nonce!")
  440. }
  441. encryptedQueryLeader := box.Seal(nonce[:], messagesFlattened[0], &nonce, leaderPublicKey, clientPrivateKey[clientNumber])
  442. _, err = rand.Read(nonce[:])
  443. if err != nil {
  444. panic("couldn't get randomness for nonce!")
  445. }
  446. encryptedQueryFollower := box.Seal(nonce[:], messagesFlattened[1], &nonce, followerPublicKey, clientPrivateKey[clientNumber])
  447. return encryptedQueryLeader, encryptedQueryFollower
  448. }
  449. //generates a topicOfInterest array with random values
  450. func addFakeInterests(max int, topicsOfInterest []int, doAuditing bool) []int {
  451. tmpNeededSubscriptions := neededSubscriptions
  452. if tmpNeededSubscriptions > len(topicList) {
  453. tmpNeededSubscriptions = len(topicList)
  454. }
  455. faketopicsOfInterest := make([]int, tmpNeededSubscriptions)
  456. maxInt := max
  457. //fills the array with unique random ascending values ranging from 0 to max
  458. for i := 0; i < tmpNeededSubscriptions; i++ {
  459. faketopicsOfInterest[i] = mr.Intn(maxInt)
  460. for j := 0; j < i; j++ {
  461. if faketopicsOfInterest[i] == faketopicsOfInterest[j] {
  462. i--
  463. break
  464. }
  465. }
  466. }
  467. if doAuditing {
  468. sort.Ints(faketopicsOfInterest)
  469. return faketopicsOfInterest
  470. }
  471. //adds unique and new random numbers to topicOfInterests until length is satisfied
  472. for _, number := range faketopicsOfInterest {
  473. if !inList(number, topicsOfInterest) {
  474. topicsOfInterest = append(topicsOfInterest, number)
  475. }
  476. if len(topicsOfInterest) == tmpNeededSubscriptions {
  477. break
  478. }
  479. }
  480. sort.Ints(topicsOfInterest)
  481. return topicsOfInterest
  482. }
  483. func inList(number int, list []int) bool {
  484. for _, element := range list {
  485. if element == number {
  486. return true
  487. }
  488. }
  489. return false
  490. }
  491. func receiveTopicLists(leaderConn net.Conn) {
  492. for i := 0; i < 2; i++ {
  493. topicListLength := readFrom(leaderConn, 4)
  494. recTopicList := readFrom(leaderConn, byteToInt(topicListLength))
  495. var tmpTopicList []string
  496. arrayReader := bytes.NewReader(recTopicList[:])
  497. json.NewDecoder(arrayReader).Decode(&tmpTopicList)
  498. if i == 0 {
  499. topicList = tmpTopicList
  500. } else {
  501. archiveTopicList = tmpTopicList
  502. }
  503. }
  504. }
  505. func getRealTweet(clientNumber int) []byte {
  506. fUserList, err := os.Open("/home/simon/goCode/tweets/userList")
  507. if err != nil {
  508. panic(err)
  509. }
  510. defer fUserList.Close()
  511. currentLine := 0
  512. scanner := bufio.NewScanner(fUserList)
  513. userID := ""
  514. for scanner.Scan() {
  515. if currentLine == clientNumber {
  516. userID = scanner.Text()
  517. break
  518. }
  519. currentLine++
  520. }
  521. if userID == "" {
  522. panic("no userID picked")
  523. }
  524. fTweets, err := os.Open("/home/simon/goCode/tweets/userTweets/" + userID)
  525. if err != nil {
  526. panic(err)
  527. }
  528. defer fTweets.Close()
  529. scanner = bufio.NewScanner(fTweets)
  530. lowerBound := timeBounds[round-1]
  531. upperBound := timeBounds[round]
  532. var tweet []byte
  533. for scanner.Scan() {
  534. //skips round 1, cause of 90% publisher rate
  535. if round == 1 {
  536. break
  537. }
  538. lineArr := strings.Split(scanner.Text(), ", \"hashtags\"")
  539. lineArr = strings.Split(lineArr[0], ": ")
  540. lineArr = strings.Split(lineArr[1], " \"")
  541. timestamp, _ := strconv.Atoi(lineArr[0])
  542. //transforms timestamp to current time
  543. timestamp -= 1351742400
  544. timestamp += startTime
  545. if float64(timestamp) > lowerBound && float64(timestamp) < upperBound {
  546. lineArr = strings.Split(scanner.Text(), "[\"")
  547. line := lineArr[1]
  548. lineArr = strings.Split(line, "\"]")
  549. line = lineArr[0]
  550. lineArr = strings.Split(line, ",")
  551. line = strings.Join(lineArr, "")
  552. topicLine := strings.Split(line, "\"")
  553. var topics []byte
  554. for index, topic := range topicLine {
  555. if index%2 == 1 {
  556. continue
  557. }
  558. if len(topics)+len(topic) > dataLength-10 {
  559. break
  560. }
  561. topics = append(topics, []byte(topic)[:]...)
  562. topics = append(topics, []byte(",")[0])
  563. }
  564. if len(topics) == 0 {
  565. break
  566. }
  567. topics = topics[:len(topics)-1]
  568. tweet = append(tweet, topics...)
  569. tweet = append(tweet, []byte(";")[0])
  570. r := mr.New(mr.NewSource(time.Now().UnixNano()))
  571. num := r.Intn(10000)
  572. if num == 0 {
  573. num = 1
  574. }
  575. tweet = append(tweet, []byte(strconv.Itoa(num) + ";;")[:]...)
  576. //adds padding
  577. length := dataLength - len(tweet)
  578. padding := make([]byte, length)
  579. rand.Read(padding)
  580. tweet = append(tweet, padding...)
  581. publisherAmount++
  582. return tweet
  583. }
  584. }
  585. tweet = make([]byte, dataLength)
  586. return tweet
  587. }
  588. func getTimeBounds() {
  589. timeBounds = make([]float64, 10000)
  590. timeBounds[0] = float64(time.Now().Unix())
  591. for index := range timeBounds {
  592. if index == 0 {
  593. continue
  594. }
  595. timeBounds[index] = timeBounds[index-1] + speedUp*(3*maxTimePerRound.Seconds()+2)
  596. }
  597. }
  598. func getRandomTweet(clientNumber int) []byte {
  599. var tweet []byte
  600. r := mr.New(mr.NewSource(time.Now().UnixNano()))
  601. maxTopics := r.Intn(6)
  602. if maxTopics == 0 {
  603. maxTopics = 1
  604. }
  605. maxInt := 100
  606. topicNumbers := make([]int, maxTopics)
  607. //fills the array with unique random ascending values ranging from 0 to maxInt
  608. for i := 0; i < maxTopics; i++ {
  609. topicNumbers[i] = mr.Intn(maxInt)
  610. for j := 0; j < i; j++ {
  611. if topicNumbers[i] == topicNumbers[j] {
  612. i--
  613. break
  614. }
  615. }
  616. }
  617. sort.Ints(topicNumbers)
  618. var topics []byte
  619. topicIndex := 0
  620. for i := 0; i < len(topicNumbers)*2; i++ {
  621. if i%2 == 0 {
  622. topics = append(topics, byte(topicNumbers[topicIndex]))
  623. topicIndex++
  624. } else if i != (len(topicNumbers)*2)-1 {
  625. topics = append(topics, []byte(",")[0])
  626. }
  627. }
  628. topics = append(topics, []byte(";")[0])
  629. num := r.Intn(100)
  630. if num == 0 {
  631. num = 1
  632. }
  633. text := []byte(strconv.Itoa(num) + ";")
  634. tweet = append(tweet, topics...)
  635. tweet = append(tweet, text...)
  636. tweet = append(tweet, []byte(";")[0])
  637. //adds padding
  638. length := dataLength - len(tweet)
  639. padding := make([]byte, length)
  640. rand.Read(padding)
  641. tweet = append(tweet, padding...)
  642. return tweet
  643. }
  644. //sends the array to the connection
  645. func writeTo(connection net.Conn, array []byte) {
  646. remainingLength := len(array)
  647. for remainingLength > 0 {
  648. if remainingLength >= mtu {
  649. _, err := connection.Write(array[:mtu])
  650. if err != nil {
  651. panic(err)
  652. }
  653. array = array[mtu:]
  654. remainingLength -= mtu
  655. } else {
  656. _, err := connection.Write(array)
  657. if err != nil {
  658. panic(err)
  659. }
  660. remainingLength = 0
  661. }
  662. }
  663. }
  664. //reads an array which is returned and of size "size" from the connection
  665. func readFrom(connection net.Conn, size int) []byte {
  666. var array []byte
  667. remainingSize := size
  668. for remainingSize > 0 {
  669. var err error
  670. toAppend := make([]byte, mtu)
  671. if remainingSize > mtu {
  672. _, err = connection.Read(toAppend)
  673. array = append(array, toAppend...)
  674. remainingSize -= mtu
  675. } else {
  676. _, err = connection.Read(toAppend[:remainingSize])
  677. array = append(array, toAppend[:remainingSize]...)
  678. remainingSize = 0
  679. }
  680. if err != nil {
  681. panic(err)
  682. }
  683. }
  684. return array
  685. }
  686. func intToByte(myInt int) (retBytes []byte) {
  687. retBytes = make([]byte, 4)
  688. retBytes[3] = byte((myInt >> 24) & 0xff)
  689. retBytes[2] = byte((myInt >> 16) & 0xff)
  690. retBytes[1] = byte((myInt >> 8) & 0xff)
  691. retBytes[0] = byte(myInt & 0xff)
  692. return
  693. }
  694. func byteToInt(myBytes []byte) (x int) {
  695. x = int(myBytes[3])<<24 + int(myBytes[2])<<16 + int(myBytes[1])<<8 + int(myBytes[0])
  696. return
  697. }