examples/java/bittorrent/Peer.java

   1 /* Copyright (c) 2006-2014. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 package bittorrent;
   8
   9 import java.util.ArrayList;
  10 import java.util.HashMap;
  11 import java.util.Iterator;
  12 import java.util.Map.Entry;
  13
  14 import org.simgrid.msg.Comm;
  15 import org.simgrid.msg.Host;
  16 import org.simgrid.msg.Msg;
  17 import org.simgrid.msg.MsgException;
  18 import org.simgrid.msg.Process;
  19 import org.simgrid.msg.RngStream;
  20 import org.simgrid.msg.Task;
  21
  22 /**
  23  * Main class for peers execution
  24  */
  25 public class Peer extends Process {
  26         protected int round = 0;
  27
  28         protected double beginReceiveTime;
  29         protected double deadline;
  30
  31         protected static RngStream stream = new RngStream();
  32
  33         protected int id;
  34         protected String mailbox;
  35         protected String mailboxTracker;
  36         protected String hostname;
  37         protected int pieces = 0;
  38         protected char[] bitfield = new char[Common.FILE_PIECES];
  39         protected char[][] bitfieldBlocks = new char[Common.FILE_PIECES][Common.PIECES_BLOCKS];
  40
  41         protected short[] piecesCount = new short[Common.FILE_PIECES];
  42
  43         protected int piecesRequested = 0;
  44
  45         protected ArrayList<Integer> currentPieces = new ArrayList<Integer>();
  46         protected int currentPiece = -1;
  47
  48         protected HashMap<Integer, Connection> activePeers = new HashMap<Integer, Connection>();
  49         protected HashMap<Integer, Connection> peers = new HashMap<Integer, Connection>();
  50
  51         protected Comm commReceived = null;
  52
  53         public Peer(Host host, String name, String[]args) {
  54                 super(host,name,args);
  55         }
  56
  57         @Override
  58         public void main(String[] args) throws MsgException {
  59                 //Check arguments
  60                 if (args.length != 3 && args.length != 2) {
  61                         Msg.info("Wrong number of arguments");
  62                 }
  63                 if (args.length == 3) {
  64                         init(Integer.valueOf(args[0]),true);
  65                 }
  66                 else {
  67                         init(Integer.valueOf(args[0]),false);
  68                 }
  69                 //Retrieve the deadline
  70                 deadline = Double.valueOf(args[1]);
  71                 if (deadline < 0) {
  72                         Msg.info("Wrong deadline supplied");
  73                         return;
  74                 }
  75                 Msg.info("Hi, I'm joining the network with id " + id);
  76                 //Getting peer data from the tracker
  77                 if (getPeersData()) {
  78                         Msg.debug("Got " + peers.size() + " peers from the tracker");
  79                         Msg.debug("Here is my current status: " + getStatus());
  80                         beginReceiveTime = Msg.getClock();
  81                         if (hasFinished()) {
  82                                 pieces = Common.FILE_PIECES;
  83                                 sendHandshakeAll();
  84                                 seedLoop();
  85                         }
  86                         else {
  87                                 leechLoop();
  88                                 seedLoop();
  89                         }
  90                 }
  91                 else {
  92                         Msg.info("Couldn't contact the tracker.");
  93                 }
  94                 Msg.info("Here is my current status: " + getStatus());
  95         }
  96         /**
  97          * Peer main loop when it is leeching.
  98          */
  99         private void leechLoop() {
 100                 double nextChokedUpdate = Msg.getClock() + Common.UPDATE_CHOKED_INTERVAL;
 101                 Msg.debug("Start downloading.");
 102                 /**
 103                  * Send a "handshake" message to all the peers it got
 104                  * (it couldn't have gotten more than 50 peers anyway)
 105                  */
 106                 sendHandshakeAll();
 107                 //Wait for at least one "bitfield" message.
 108                 waitForPieces();
 109                 Msg.debug("Starting main leech loop");
 110                 while (Msg.getClock() < deadline && pieces < Common.FILE_PIECES) {
 111                         if (commReceived == null) {
 112                                 commReceived = Task.irecv(mailbox);
 113                         }
 114                         try {
 115                                 if (commReceived.test()) {
 116                                         handleMessage(commReceived.getTask());
 117                                         commReceived = null;
 118                                 }
 119                                 else {
 120                                         //If the user has a pending interesting
 121                                         if (currentPiece != -1) {
 122                                                 sendInterestedToPeers();
 123                                         }
 124                                         else {
 125                                                 if (currentPieces.size() < Common.MAX_PIECES) {
 126                                                         updateCurrentPiece();
 127                                                 }
 128                                         }
 129                                         //We don't execute the choke algorithm if we don't already have a piece
 130                                         if (Msg.getClock() >= nextChokedUpdate && pieces > 0) {
 131                                                 updateChokedPeers();
 132                                                 nextChokedUpdate += Common.UPDATE_CHOKED_INTERVAL;
 133                                         }
 134                                         else {
 135                                                 waitFor(1);
 136                                         }
 137                                 }
 138                         }
 139                         catch (MsgException e) {
 140                                 commReceived = null;
 141                         }
 142                 }
 143         }
 144
 145         /**
 146          * Peer main loop when it is seeding
 147          */
 148         private void seedLoop() {
 149                 double nextChokedUpdate = Msg.getClock() + Common.UPDATE_CHOKED_INTERVAL;
 150                 Msg.debug("Start seeding.");
 151                 //start the main seed loop
 152                 while (Msg.getClock() < deadline) {
 153                         if (commReceived == null) {
 154                                 commReceived = Task.irecv(mailbox);
 155                         }
 156                         try {
 157                                 if (commReceived.test()) {
 158                                         handleMessage(commReceived.getTask());
 159                                         commReceived = null;
 160                                 }
 161                                 else {
 162                                         if (Msg.getClock() >= nextChokedUpdate) {
 163                                                 updateChokedPeers();
 164                                                 //TODO: Change the choked peer algorithm when seeding
 165                                                 nextChokedUpdate += Common.UPDATE_CHOKED_INTERVAL;
 166                                         }
 167                                         else {
 168                                                 waitFor(1);
 169                                         }
 170                                 }
 171                         }
 172                         catch (MsgException e) {
 173                                 commReceived = null;
 174                         }
 175
 176                 }
 177         }
 178
 179         /**
 180          * Initialize the various peer data
 181          * @param id id of the peer to take in the network
 182          * @param seed indicates if the peer is a seed
 183          */
 184         private void init(int id, boolean seed) {
 185                 this.id = id;
 186                 this.mailbox = Integer.toString(id);
 187                 this.mailboxTracker = "tracker_" + Integer.toString(id);
 188                 if (seed) {
 189                         for (int i = 0; i < bitfield.length; i++) {
 190                                 bitfield[i] = '1';
 191                                 for (int j = 0; j < bitfieldBlocks[i].length; j++) {
 192                                         bitfieldBlocks[i][j] = '1';
 193                                 }
 194                         }
 195                 }
 196                 else {
 197                         for (int i = 0; i < bitfield.length; i++) {
 198                                 bitfield[i] = '0';
 199                                 for (int j = 0; j < bitfieldBlocks[i].length; j++) {
 200                                         bitfieldBlocks[i][j] = '0'      ;
 201                                 }
 202                         }
 203                 }
 204                 this.hostname = getHost().getName();
 205         }
 206         /**
 207          * Retrieves the peer list from the tracker
 208          */
 209         private boolean getPeersData() {
 210
 211                 boolean success = false, sendSuccess = false;
 212                 double timeout = Msg.getClock() + Common.GET_PEERS_TIMEOUT;
 213                 //Build the task to send to the tracker
 214                 TrackerTask taskSend = new TrackerTask(hostname, mailboxTracker, id);
 215
 216                 while (!sendSuccess && Msg.getClock() < timeout) {
 217                         try {
 218                                 Msg.debug("Sending a peer request to the tracker.");
 219                                 taskSend.send(Common.TRACKER_MAILBOX,Common.GET_PEERS_TIMEOUT);
 220                                 sendSuccess = true;
 221                         }
 222                         catch (MsgException e) {
 223
 224                         }
 225                 }
 226                 while (!success && Msg.getClock() < timeout) {
 227                         commReceived = Task.irecv(this.mailboxTracker);
 228                         try {
 229                                 commReceived.waitCompletion(Common.GET_PEERS_TIMEOUT);
 230                                 if (commReceived.getTask() instanceof TrackerTask) {
 231                                         TrackerTask task = (TrackerTask)commReceived.getTask();
 232                                         for (Integer peerId: task.peers) {
 233                                                 if (peerId != this.id) {
 234                                                         peers.put(peerId, new Connection(peerId));
 235                                                 }
 236                                         }
 237                                         success = true;
 238                                 }
 239                         }
 240                         catch (MsgException e) {
 241
 242                         }
 243                         commReceived = null;
 244                 }
 245                 commReceived = null;
 246                 return success;
 247         }
 248         /**
 249          * Handle a received message sent by another peer
 250          * @param task task received.
 251          */
 252         void handleMessage(Task task) {
 253                 MessageTask message = (MessageTask)task;
 254                 Connection remotePeer = peers.get(message.peerId);
 255                 switch (message.type) {
 256                         case HANDSHAKE:
 257                                 Msg.debug("Received a HANDSHAKE message from " + message.mailbox);
 258                                 //Check if the peer is in our connection list
 259                                 if (remotePeer == null) {
 260                                         peers.put(message.peerId, new Connection(message.peerId));
 261                                         sendHandshake(message.mailbox);
 262                                 }
 263                                 //Send our bitfield to the pair
 264                                 sendBitfield(message.mailbox);
 265                         break;
 266                         case BITFIELD:
 267                                 Msg.debug("Received a BITFIELD message from " + message.peerId + " (" + message.issuerHostname + ")");
 268                                 //update the pieces list
 269                                 updatePiecesCountFromBitfield(message.bitfield);
 270                                 //Update the current piece
 271                                 if (currentPiece == -1 && pieces < Common.FILE_PIECES && currentPieces.size() < Common.MAX_PIECES) {
 272                                         updateCurrentPiece();
 273                                 }
 274                                 remotePeer.bitfield  = message.bitfield.clone();
 275                         break;
 276                         case INTERESTED:
 277                                 Msg.debug("Received an INTERESTED message from " + message.peerId + " (" + message.issuerHostname + ")");
 278                                 assert remotePeer != null;
 279                                 remotePeer.interested = true;
 280                         break;
 281                         case NOTINTERESTED:
 282                                 Msg.debug("Received a NOTINTERESTED message from " + message.peerId + " (" + message.issuerHostname + ")");
 283                                 assert remotePeer != null;
 284                                 remotePeer.interested = false;
 285                         break;
 286                         case UNCHOKE:
 287                                 Msg.debug("Received an UNCHOKE message from " + message.peerId + "(" + message.issuerHostname + ")");
 288                                 assert remotePeer != null;
 289                                 remotePeer.chokedDownload = false;
 290                                 activePeers.put(remotePeer.id,remotePeer);
 291                                 sendRequestsToPeer(remotePeer);
 292                         break;
 293                         case CHOKE:
 294                                 Msg.debug("Received a CHOKE message from " + message.peerId + " (" + message.issuerHostname + ")");
 295                                 assert remotePeer != null;
 296                                 remotePeer.chokedDownload = true;
 297                                 activePeers.remove(remotePeer.id);
 298                         break;
 299                         case HAVE:
 300                                 if (remotePeer.bitfield == null) {
 301                                         return;
 302                                 }
 303                                 Msg.debug("Received a HAVE message from " + message.peerId + " (" + message.issuerHostname + ")");
 304                                 assert message.index >= 0 && message.index < Common.FILE_PIECES;
 305                                 assert remotePeer.bitfield != null;
 306                                 remotePeer.bitfield[message.index] = '1';
 307                                 piecesCount[message.index]++;
 308                                 //Send interested message to the peer if he has what we want
 309                                 if (!remotePeer.amInterested && currentPieces.contains(message.index) ) {
 310                                         remotePeer.amInterested = true;
 311                                         sendInterested(remotePeer.mailbox);
 312                                 }
 313
 314                                 if (currentPieces.contains(message.index)) {
 315                                         int blockIndex = getFirstBlock(message.index);
 316                                         int blockLength = Common.PIECES_BLOCKS - blockIndex ;
 317                                         blockLength = blockLength > Common.BLOCKS_REQUESTED ? Common.BLOCKS_REQUESTED : blockLength;
 318                                         sendRequest(message.mailbox,message.index,blockIndex,blockLength);
 319                                 }
 320                         break;
 321                         case REQUEST:
 322                                 assert message.index >= 0 && message.index < Common.FILE_PIECES;
 323                                 if (!remotePeer.chokedUpload) {
 324                                         Msg.debug("Received a REQUEST from " + message.peerId + "(" + message.issuerHostname + ") for " + message.peerId);
 325                                         if (bitfield[message.index] == '1') {
 326                                                 sendPiece(message.mailbox,message.index,false,message.blockIndex,message.blockLength);
 327                                         }
 328                                         else {
 329                                                 Msg.debug("Received a REQUEST from " + message.peerId + " (" + message.issuerHostname + ") but he is choked" );
 330                                         }
 331                                 }
 332                         break;
 333                         case PIECE:
 334                                 if (message.stalled) {
 335                                         Msg.debug("The received piece " + message.index + " from " + message.peerId + " (" + message.issuerHostname + ") is stalled");
 336                                 }
 337                                 else {
 338                                         Msg.debug("Received piece " + message.index + " from " + message.peerId + " (" + message.issuerHostname + ")");
 339                                         if (bitfield[message.index] == '0') {
 340                                                 updateBitfieldBlocks(message.index,message.blockIndex,message.blockLength);
 341                                                 if (pieceComplete(message.index)) {
 342                                                         piecesRequested--;
 343                                                         //Removing the piece from our piece list.
 344                                                         if (!currentPieces.remove((Object)Integer.valueOf(message.index))) {
 345                                                         }
 346                                                         //Setting the fact that we have the piece
 347                                                         bitfield[message.index] = '1';
 348                                                         pieces++;
 349                                                         Msg.debug("My status is now " + getStatus());
 350                                                         //Sending the information to all the peers we are connected to
 351                                                         sendHave(message.index);
 352                                                         //sending UNINTERESTED to peers that doesn't have what we want.
 353                                                         updateInterestedAfterReceive();
 354                                                 }
 355                                         }
 356                                         else {
 357                                                 Msg.debug("However, we already have it.");
 358                                         }
 359                                 }
 360                         break;
 361                 }
 362                 if (remotePeer != null) {
 363                         remotePeer.addSpeedValue(1 / (Msg.getClock() - beginReceiveTime));
 364                 }
 365                 beginReceiveTime = Msg.getClock();
 366         }
 367         /**
 368          * Wait for the node to receive interesting bitfield messages (ie: non empty)
 369          * to be received
 370          */
 371         void waitForPieces() {
 372                 boolean finished = false;
 373                 while (Msg.getClock() < deadline && !finished) {
 374                         if (commReceived == null) {
 375                                 commReceived = Task.irecv(mailbox);
 376                         }
 377                         try {
 378                                 commReceived.waitCompletion(Common.TIMEOUT_MESSAGE);
 379                                 handleMessage(commReceived.getTask());
 380                                 if (currentPiece != -1) {
 381                                         finished = true;
 382                                 }
 383                                 commReceived = null;
 384                         }
 385                         catch (MsgException e) {
 386                                 commReceived = null;
 387                         }
 388                 }
 389         }
 390
 391         private boolean hasFinished() {
 392                 for (int i = 0; i < bitfield.length; i++) {
 393                         if (bitfield[i] == '1') {
 394                                 return true;
 395                         }
 396                 }
 397                 return false;
 398         }
 399         /**
 400          * Updates the list of who has a piece from a bitfield
 401          * @param bitfield bitfield
 402          */
 403         private void updatePiecesCountFromBitfield(char bitfield[]) {
 404                 for (int i = 0; i < Common.FILE_PIECES; i++) {
 405                         if (bitfield[i] == '1') {
 406                                 piecesCount[i]++;
 407                         }
 408                 }
 409         }
 410         /**
 411          * Update the piece the peer is currently interested in.
 412          * There is two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
 413          * If the peer has less than 3 pieces, he chooses a piece at random.
 414          * If the peer has more than pieces, he downloads the pieces that are the less
 415          * replicated
 416          */
 417         void updateCurrentPiece() {
 418                 if (currentPieces.size() >= (Common.FILE_PIECES - pieces)) {
 419                         return;
 420                 }
 421                 if (true || pieces < 3) {
 422                         int peerPiece;
 423                         do {
 424                                 currentPiece = stream.randInt(0,Common.FILE_PIECES - 1);
 425                         } while (!(bitfield[currentPiece] == '0' && !currentPieces.contains(currentPiece)));
 426                 }
 427                 else {
 428                         //trivial min algorithm.
 429                         //TODO
 430                 }
 431                 currentPieces.add(currentPiece);
 432                 Msg.debug("New interested piece: " + currentPiece);
 433                 assert currentPiece >= 0 && currentPiece < Common.FILE_PIECES;
 434         }
 435         /**
 436          * Update the list of current choked and unchoked peers, using the
 437          * choke algorithm
 438          */
 439         private void updateChokedPeers() {
 440                 round = (round + 1) % 3;
 441                 if (peers.size() == 0) {
 442                         return;
 443                 }
 444                 //remove a peer from the list
 445                 Iterator<Entry<Integer, Connection>> it = activePeers.entrySet().iterator();
 446                 if (it.hasNext()) {
 447                         Entry<Integer,Connection> e = it.next();
 448                         Connection peerChoked = e.getValue();
 449                         peerChoked.chokedUpload = true;
 450                         sendChoked(peerChoked.mailbox);
 451                         activePeers.remove(e.getKey());
 452                 }
 453                 Connection peerChoosed = null;
 454                 //Separate the case from when the peer is seeding.
 455                 if (pieces == Common.FILE_PIECES) {
 456                         //Find the last unchoked peer.
 457                         double unchokeTime = deadline + 1;
 458                         for (Connection connection : peers.values()) {
 459                                 if (connection.lastUnchoke < unchokeTime && connection.interested) {
 460                                         peerChoosed = connection;
 461                                         unchokeTime = connection.lastUnchoke;
 462                                 }
 463                         }
 464                 }
 465                 else {
 466                         //Random optimistic unchoking
 467                         if (round == 0) {
 468                                 int j = 0, i;
 469                                 do {
 470                                         i = 0;
 471                                         int idChosen = stream.randInt(0,peers.size() - 1);
 472                                         for (Connection connection : peers.values()) {
 473                                                 if (i == idChosen) {
 474                                                         peerChoosed = connection;
 475                                                         break;
 476                                                 }
 477                                                 i++;
 478                                         } //TODO: Not really the best way ever
 479                                         if (!peerChoosed.interested) {
 480                                                 peerChoosed = null;
 481                                         }
 482                                         j++;
 483                                 } while (peerChoosed == null && j < Common.MAXIMUM_PEERS);
 484                         }
 485                         else {
 486                                 Connection fastest = null;
 487                                 double fastestSpeed = 0;
 488                                 for (Connection c : peers.values()) {
 489                                         if (c.peerSpeed > fastestSpeed && c.interested && c.chokedUpload) {
 490                                                 fastest = c;
 491                                                 fastestSpeed = c.peerSpeed;
 492                                         }
 493                                 }
 494                                 peerChoosed = fastest;
 495                         }
 496                 }
 497                 if (peerChoosed != null) {
 498                         activePeers.put(peerChoosed.id,peerChoosed);
 499                         peerChoosed.chokedUpload = false;
 500                         peerChoosed.lastUnchoke = Msg.getClock();
 501                         sendUnchoked(peerChoosed.mailbox);
 502                 }
 503         }
 504         /**
 505          * Updates our "interested" state about peers: send "not interested" to peers
 506          * that don't have any more pieces we want.
 507          */
 508         private void updateInterestedAfterReceive() {
 509                 boolean interested;
 510                 for (Connection connection : peers.values()) {
 511                         interested = false;
 512                         if (connection.amInterested) {
 513                                 for (Integer piece : currentPieces) {
 514                                         if (connection.bitfield[piece] == '1') {
 515                                                 interested = true;
 516                                                 break;
 517                                         }
 518                                 }
 519                                 if (!interested) {
 520                                         connection.amInterested = false;
 521                                         sendNotInterested(connection.mailbox);
 522                                 }
 523                         }
 524                 }
 525         }
 526         private void updateBitfieldBlocks(int index, int blockIndex, int blockLength) {
 527                 for (int i = blockIndex; i < (blockIndex + blockLength); i++) {
 528                         bitfieldBlocks[index][i] = '1';
 529                 }
 530         }
 531         /**
 532          * Returns if a piece is complete in the peer's bitfield.
 533          * @param index the index of the piece.
 534          */
 535         private boolean pieceComplete(int index) {
 536                 for (int i = 0; i < bitfieldBlocks[index].length; i++) {
 537                         if (bitfieldBlocks[index][i] == '0') {
 538                                 return false;
 539                         }
 540                 }
 541                 return true;
 542         }
 543         /**
 544          * Returns the first block of a piece that we don't have.
 545          */
 546         private int getFirstBlock(int piece) {
 547                 int blockIndex = -1;
 548                 for (int i = 0; i < Common.PIECES_BLOCKS; i++) {
 549                         if (bitfieldBlocks[piece][i] == '0') {
 550                                 blockIndex = i;
 551                                 break;
 552                         }
 553                 }
 554                 return blockIndex;
 555         }
 556         /**
 557          * Send request messages to a peer that have unchoked us
 558          * @param remotePeer peer data to the peer we want to send the request
 559          */
 560         private void sendRequestsToPeer(Connection remotePeer) {
 561                 if (remotePeer.bitfield == null) {
 562                         return;
 563                 }
 564                 for (Integer piece : currentPieces) {
 565                         //Getting the block to send.
 566                         int blockIndex = -1, blockLength = 0;
 567                         blockIndex = getFirstBlock(piece);
 568                         blockLength = Common.PIECES_BLOCKS - blockIndex ;
 569                         blockLength = blockLength > Common.BLOCKS_REQUESTED ? Common.BLOCKS_REQUESTED : blockLength;
 570                         if (remotePeer.bitfield[piece] == '1') {
 571                                 sendRequest(remotePeer.mailbox, piece, blockIndex, blockLength);
 572                         }
 573                 }
 574         }
 575         /**
 576          * Find the peers that have the current interested piece and send them
 577          * the "interested" message
 578          */
 579         private void sendInterestedToPeers() {
 580                 if (currentPiece == -1) {
 581                         return;
 582                 }
 583                 for (Connection connection : peers.values()) {
 584                         if (connection.bitfield != null && connection.bitfield[currentPiece] == '1' && !connection.amInterested) {
 585                                 connection.amInterested = true;
 586                                 MessageTask task = new MessageTask(MessageTask.Type.INTERESTED, hostname, this.mailbox, this.id);
 587                                 task.dsend(connection.mailbox);
 588                         }
 589                 }
 590                 currentPiece = -1;
 591                 piecesRequested++;
 592         }
 593         /**
 594          * Send a "interested" message to a peer.
 595          */
 596         private void sendInterested(String mailbox) {
 597                 MessageTask task = new MessageTask(MessageTask.Type.INTERESTED, hostname, this.mailbox, this.id);
 598                 task.dsend(mailbox);
 599         }
 600         /**
 601          * Send a "not interested" message to a peer
 602          * @param mailbox mailbox destination mailbox
 603          */
 604         private void sendNotInterested(String mailbox) {
 605                 MessageTask task = new MessageTask(MessageTask.Type.NOTINTERESTED, hostname, this.mailbox, this.id);
 606                 task.dsend(mailbox);
 607         }
 608         /**
 609          * Send a handshake message to all the peers the peer has.
 610          * @param peer peer data
 611          */
 612         private void sendHandshakeAll() {
 613                 for (Connection remotePeer : peers.values()) {
 614                         MessageTask task = new MessageTask(MessageTask.Type.HANDSHAKE, hostname, mailbox,
 615                         id);
 616                         task.dsend(remotePeer.mailbox);
 617                 }
 618         }
 619         /**
 620          * Send a "handshake" message to an user
 621          * @param mailbox mailbox where to we send the message
 622          */
 623         private void sendHandshake(String mailbox) {
 624                 Msg.debug("Sending a HANDSHAKE to " + mailbox);
 625                 MessageTask task = new MessageTask(MessageTask.Type.HANDSHAKE, hostname, this.mailbox, this.id);
 626                 task.dsend(mailbox);
 627         }
 628         /**
 629          * Send a "choked" message to a peer
 630          */
 631         private void sendChoked(String mailbox) {
 632                 Msg.debug("Sending a CHOKE to " + mailbox);
 633                 MessageTask task = new MessageTask(MessageTask.Type.CHOKE, hostname, this.mailbox, this.id);
 634                 task.dsend(mailbox);
 635         }
 636         /**
 637          * Send a "unchoked" message to a peer
 638          */
 639         private void sendUnchoked(String mailbox) {
 640                 Msg.debug("Sending a UNCHOKE to " + mailbox);
 641                 MessageTask task = new MessageTask(MessageTask.Type.UNCHOKE, hostname, this.mailbox, this.id);
 642                 task.dsend(mailbox);
 643         }
 644         /**
 645          * Send a "HAVE" message to all peers we are connected to
 646          */
 647         private void sendHave(int piece) {
 648                 Msg.debug("Sending HAVE message to all my peers");
 649                 for (Connection remotePeer : peers.values()) {
 650                         MessageTask task = new MessageTask(MessageTask.Type.HAVE, hostname, this.mailbox, this.id, piece);
 651                         task.dsend(remotePeer.mailbox);
 652                 }
 653         }
 654         /**
 655          * Send a bitfield message to all the peers the peer has.
 656          * @param peer peer data
 657          */
 658         private void sendBitfield(String mailbox) {
 659                 Msg.debug("Sending a BITFIELD to " + mailbox);
 660                 MessageTask task = new MessageTask(MessageTask.Type.BITFIELD, hostname, this.mailbox, this.id, this.bitfield);
 661                 task.dsend(mailbox);
 662         }
 663         /**
 664          * Send a "request" message to a pair, containing a request for a piece
 665          */
 666         private void sendRequest(String mailbox, int piece, int blockIndex, int blockLength) {
 667                 Msg.debug("Sending a REQUEST to " + mailbox + " for piece " + piece + " and blocks " + blockIndex + "," + (blockIndex + blockLength));
 668                 MessageTask task = new MessageTask(MessageTask.Type.REQUEST, hostname, this.mailbox, this.id, piece, blockIndex, blockLength);
 669                 task.dsend(mailbox);
 670         }
 671         /**
 672          * Send a "piece" message to a pair, containing a piece of the file
 673          */
 674         private void sendPiece(String mailbox, int piece, boolean stalled, int blockIndex, int blockLength) {
 675                 Msg.debug("Sending the PIECE " + piece + " to " + mailbox);
 676                 MessageTask task = new MessageTask(MessageTask.Type.PIECE, hostname, this.mailbox, this.id, piece, stalled, blockIndex, blockLength);
 677                 task.dsend(mailbox);
 678         }
 679
 680         private String getStatus() {
 681                 String s = "";
 682                 for (int i = 0; i < Common.FILE_PIECES; i++) {
 683                         s = s + bitfield[i];
 684                 }
 685                 return s;
 686         }
 687 }
 688