<commit_before>/* PeerConnectionOut - Keeps a queue of outgoing messages and delivers them.
   Copyright (C) 2003 <NAME>

   This file is part of Snark.
   
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.
 
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
 
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software Foundation,
   Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*/

package org.klomp.snark;

import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import net.i2p.util.I2PAppThread;
import net.i2p.util.Log;
import net.i2p.util.SimpleScheduler;
import net.i2p.util.SimpleTimer;

class PeerConnectionOut implements Runnable
{
  private Log _log = new Log(PeerConnectionOut.class);
  private final Peer peer;
  private final DataOutputStream dout;

  private Thread thread;
  private boolean quit;

  // Contains Messages.
  private final List sendQueue = new ArrayList();
  
  private static long __id = 0;
  private long _id;
  
  long lastSent;

  public PeerConnectionOut(Peer peer, DataOutputStream dout)
  {
    this.peer = peer;
    this.dout = dout;
    _id = ++__id;

    lastSent = System.currentTimeMillis();
    quit = false;
  }
  
  public void startup() {
    thread = new I2PAppThread(this, "Snark sender " + _id + ": " + peer);
    thread.start();
  }

  /**
   * Continuesly monitors for more outgoing messages that have to be send.
   * Stops if quit is true of an IOException occurs.
   */
  public void run()
  {
    try
      {
        while (!quit && peer.isConnected())
          {
            Message m = null;
            PeerState state = null;
            boolean shouldFlush;
            synchronized(sendQueue)
              {
                shouldFlush = !quit && peer.isConnected() && sendQueue.isEmpty();
              }
            if (shouldFlush)
                // Make sure everything will reach the other side.
                // flush while not holding lock, could take a long time
                dout.flush();

            synchronized(sendQueue)
              {
                while (!quit && peer.isConnected() && sendQueue.isEmpty())
                  {
                    try
                      {
                        // Make sure everything will reach the other side.
                        // don't flush while holding lock, could take a long time
                        // dout.flush();
                        
                        // Wait till more data arrives.
                        sendQueue.wait(60*1000);
                      }
                    catch (InterruptedException ie)
                      {
                        /* ignored */
                      }
                  }
                state = peer.state;
                if (!quit && state != null && peer.isConnected())
                  {
                    // Piece messages are big. So if there are other
                    // (control) messages make sure they are send first.
                    // Also remove request messages from the queue if
                    // we are currently being choked to prevent them from
                    // being send even if we get unchoked a little later.
                    // (Since we will resent them anyway in that case.)
                    // And remove piece messages if we are choking.
                    
                    // this should get fixed for starvation
                    Iterator it = sendQueue.iterator();
                    while (m == null && it.hasNext())
                      {
                        Message nm = (Message)it.next();
                        if (nm.type == Message.PIECE)
                          {
                            if (state.choking) {
                              it.remove();
                              SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            }
                            nm = null;
                          }
                        else if (nm.type == Message.REQUEST && state.choked)
                          {
                            it.remove();
                            SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            nm = null;
                          }
                          
                        if (m == null && nm != null)
                          {
                            m = nm;
                            SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            it.remove();
                          }
                      }
                    if (m == null && sendQueue.size() > 0) {
                      m = (Message)sendQueue.remove(0);
                      SimpleTimer.getInstance().removeEvent(m.expireEvent);
                    }
                  }
              }
            if (m != null)
              {
                if (_log.shouldLog(Log.DEBUG))
                    _log.debug("Send " + peer + ": " + m + " on " + peer.metainfo.getName());
                m.sendMessage(dout);
                lastSent = System.currentTimeMillis();

                // Remove all piece messages after sending a choke message.
                if (m.type == Message.CHOKE)
                  removeMessage(Message.PIECE);

                // XXX - Should also register overhead...
                if (m.type == Message.PIECE)
                  state.uploaded(m.len);

                m = null;
              }<commit_msg>    * i2psnark: Better track outgoing bandwidth by incrementing       counter before the blocking write<commit_after>/* PeerConnectionOut - Keeps a queue of outgoing messages and delivers them.
   Copyright (C) 2003 <NAME>

   This file is part of Snark.
   
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.
 
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
 
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software Foundation,
   Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*/

package org.klomp.snark;

import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import net.i2p.util.I2PAppThread;
import net.i2p.util.Log;
import net.i2p.util.SimpleScheduler;
import net.i2p.util.SimpleTimer;

class PeerConnectionOut implements Runnable
{
  private Log _log = new Log(PeerConnectionOut.class);
  private final Peer peer;
  private final DataOutputStream dout;

  private Thread thread;
  private boolean quit;

  // Contains Messages.
  private final List sendQueue = new ArrayList();
  
  private static long __id = 0;
  private long _id;
  
  long lastSent;

  public PeerConnectionOut(Peer peer, DataOutputStream dout)
  {
    this.peer = peer;
    this.dout = dout;
    _id = ++__id;

    lastSent = System.currentTimeMillis();
    quit = false;
  }
  
  public void startup() {
    thread = new I2PAppThread(this, "Snark sender " + _id + ": " + peer);
    thread.start();
  }

  /**
   * Continuesly monitors for more outgoing messages that have to be send.
   * Stops if quit is true of an IOException occurs.
   */
  public void run()
  {
    try
      {
        while (!quit && peer.isConnected())
          {
            Message m = null;
            PeerState state = null;
            boolean shouldFlush;
            synchronized(sendQueue)
              {
                shouldFlush = !quit && peer.isConnected() && sendQueue.isEmpty();
              }
            if (shouldFlush)
                // Make sure everything will reach the other side.
                // flush while not holding lock, could take a long time
                dout.flush();

            synchronized(sendQueue)
              {
                while (!quit && peer.isConnected() && sendQueue.isEmpty())
                  {
                    try
                      {
                        // Make sure everything will reach the other side.
                        // don't flush while holding lock, could take a long time
                        // dout.flush();
                        
                        // Wait till more data arrives.
                        sendQueue.wait(60*1000);
                      }
                    catch (InterruptedException ie)
                      {
                        /* ignored */
                      }
                  }
                state = peer.state;
                if (!quit && state != null && peer.isConnected())
                  {
                    // Piece messages are big. So if there are other
                    // (control) messages make sure they are send first.
                    // Also remove request messages from the queue if
                    // we are currently being choked to prevent them from
                    // being send even if we get unchoked a little later.
                    // (Since we will resent them anyway in that case.)
                    // And remove piece messages if we are choking.
                    
                    // this should get fixed for starvation
                    Iterator it = sendQueue.iterator();
                    while (m == null && it.hasNext())
                      {
                        Message nm = (Message)it.next();
                        if (nm.type == Message.PIECE)
                          {
                            if (state.choking) {
                              it.remove();
                              SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            }
                            nm = null;
                          }
                        else if (nm.type == Message.REQUEST && state.choked)
                          {
                            it.remove();
                            SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            nm = null;
                          }
                          
                        if (m == null && nm != null)
                          {
                            m = nm;
                            SimpleTimer.getInstance().removeEvent(nm.expireEvent);
                            it.remove();
                          }
                      }
                    if (m == null && sendQueue.size() > 0) {
                      m = (Message)sendQueue.remove(0);
                      SimpleTimer.getInstance().removeEvent(m.expireEvent);
                    }
                  }
              }
            if (m != null)
              {
                if (_log.shouldLog(Log.DEBUG))
                    _log.debug("Send " + peer + ": " + m + " on " + peer.metainfo.getName());

                // This can block for quite a while.
                // To help get slow peers going, and track the bandwidth better,
                // move this _after_ state.uploaded() and see how it works.
                //m.sendMessage(dout);
                lastSent = System.currentTimeMillis();

                // Remove all piece messages after sending a choke message.
                if (m.type == Message.CHOKE)
                  removeMessage(Message.PIECE);

                // XXX - Should also register overhead...
                if (m.type == Message.PIECE)
                  state.uploaded(m.len);

                m.sendMessage(dout);
                m = null;
              }
{'fraction_non_alphanumeric': 0.07973718364698248, 'fraction_numerical': 0.005597014925373134, 'mean_word_length': 1.2453105081041704, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '13767', 'n_tokens_mistral': 3068, 'n_tokens_neox': 2838, 'n_words': 980}