<commit_before>//
// redis-sharp.cs: ECMA CLI Binding to the Redis key-value storage system
//
// Authors:
//   <NAME> (<EMAIL>)
//
// Copyright 2010 Novell, Inc.
//
// Licensed under the same terms of Redis: new BSD license.
//

using System;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using ServiceStack.Text;

namespace ServiceStack.Redis
{
	public partial class RedisNativeClient
	{
		private void Connect()
		{
			socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)
			{
				SendTimeout = SendTimeout,
				ReceiveTimeout = ReceiveTimeout
			};
			try
			{
                if (ConnectTimeout == 0)
                {
                    socket.Connect(Host, Port);
                }
                else
                {
                    var connectResult = socket.BeginConnect(Host, Port, null, null);
                    connectResult.AsyncWaitHandle.WaitOne(ConnectTimeout, true);
                }

				if (!socket.Connected)
				{
					socket.Close();
					socket = null;
					return;
				}
				Bstream = new BufferedStream(new NetworkStream(socket), 16 * 1024);

				if (Password != null)
					SendExpectSuccess(Commands.Auth, Password.ToUtf8Bytes());

				db = 0;
				var ipEndpoint = socket.LocalEndPoint as IPEndPoint;
				clientPort = ipEndpoint != null ? ipEndpoint.Port : -1;
				lastCommand = null;
				lastSocketException = null;
				LastConnectedAtTimestamp = Stopwatch.GetTimestamp();

                if (ConnectionFilter != null)
                {
                    ConnectionFilter(this);
                }
			}
			catch (SocketException ex)
			{
                if (socket != null)
                    socket.Close();
                socket = null;

				HadExceptions = true;
				var throwEx = new RedisException("could not connect to redis Instance at " + Host + ":" + Port, ex);
				log.Error(throwEx.Message, ex);
				throw throwEx;
			}
		}

		protected string ReadLine()
		{
			var sb = new StringBuilder();

			int c;
			while ((c = Bstream.ReadByte()) != -1)
			{
				if (c == '\r')
					continue;
				if (c == '\n')
					break;
				sb.Append((char)c);
			}
			return sb.ToString();
		}

		private bool AssertConnectedSocket()
		{
			if (LastConnectedAtTimestamp > 0)
			{
				var now = Stopwatch.GetTimestamp();
				var elapsedSecs = (now - LastConnectedAtTimestamp) / Stopwatch.Frequency;

				if (socket == null || (elapsedSecs > IdleTimeOutSecs && !socket.IsConnected()))
				{
					return Reconnect();
				}
				LastConnectedAtTimestamp = now;
			}

			if (socket == null)
			{
				var previousDb = db;
				Connect();
				if (previousDb != DefaultDb) this.Db = previousDb;
			}

			var isConnected = socket != null;

			return isConnected;
		}

		private bool Reconnect()
		{
			var previousDb = db;

			SafeConnectionClose();
			Connect(); //sets db to 0

			if (previousDb != DefaultDb) this.Db = previousDb;

			return socket != null;
		}

		private bool HandleSocketException(SocketException ex)
		{
			HadExceptions = true;
			log.Error("SocketException: ", ex);

			lastSocketException = ex;

			// timeout?
			socket.Close();
			socket = null;

			return false;
		}

		private RedisResponseException CreateResponseError(string error)
		{
			HadExceptions = true;
			var throwEx = new RedisResponseException(
				string.Format("{0}, sPort: {1}, LastCommand: {2}",
					error, clientPort, lastCommand));
			log.Error(throwEx.Message);
			throw throwEx;
		}

		private RedisException CreateConnectionError()
		{
			HadExceptions = true;
            var throwEx = new RedisException(
				string.Format("Unable to Connect: sPort: {0}",
					clientPort), lastSocketException);
			log.Error(throwEx.Message);
			throw throwEx;
		}

		private static byte[] GetCmdBytes(char cmdPrefix, int noOfLines)
		{
			var strLines = noOfLines.ToString();
			var strLinesLength = strLines.Length;

			var cmdBytes = new byte[1 + strLinesLength + 2];
			cmdBytes[0] = (byte)cmdPrefix;

			for (var i = 0; i < strLinesLength; i++)
				cmdBytes[i + 1] = (byte)strLines[i];

			cmdBytes[1 + strLinesLength] = 0x0D; // \r
			cmdBytes[2 + strLinesLength] = 0x0A; // \n

			return cmdBytes;
		}

		/// <summary>
		/// Command to set multuple binary safe arguments
		/// </summary>
		/// <param name="cmdWithBinaryArgs"></param>
		/// <returns></returns>
		protected bool SendCommand(params byte[][] cmdWithBinaryArgs)
		{
			if (!AssertConnectedSocket()) return false;

			try
			{
				CmdLog(cmdWithBinaryArgs);

				//Total command lines count
				WriteAllToSendBuffer(cmdWithBinaryArgs);

                //pipeline will handle flush, if pipelining is turned on
                if (Pipeline == null)
                    FlushSendBuffer();
			}
			catch (SocketException ex)
			{
				cmdBufferIndex = 0;
				return HandleSocketException(ex);
			}
			return true;
		}

		public void WriteAllToSendBuffer(params byte[][] cmdWithBinaryArgs)
		{
			WriteToSendBuffer(GetCmdBytes('*', cmdWithBinaryArgs.Length));

			foreach (var safeBinaryValue in cmdWithBinaryArgs)
			{
				WriteToSendBuffer(GetCmdBytes('$', safeBinaryValue.Length));
				WriteToSendBuffer(safeBinaryValue);
				WriteToSendBuffer(endData);
			}
		}

		byte[] cmdBuffer = new byte[32 * 1024];
		int cmdBufferIndex = 0;

		public void WriteToSendBuffer(byte[] cmdBytes)
		{
			if ((cmdBufferIndex + cmdBytes.Length) > cmdBuffer.Length)
			{
				int requiredLength = cmdBufferIndex + cmdBytes.Length;
				const int lohThreshold = 85000 - 1;
				const int breathingSpaceToReduceReallocations = (32 * 1024);
				int newSize = lohThreshold;
				if (requiredLength <= lohThreshold) {
					if (requiredLength + breathingSpaceToReduceReallocations <= lohThreshold) {
						newSize = requiredLength + breathingSpaceToReduceReallocations;
					}
				} else {
					newSize = requiredLength + breathingSpaceToReduceReallocations;
				}
				var newLargerBuffer = new byte[newSize];
				Buffer.BlockCopy(cmdBuffer, 0, newLargerBuffer, 0, cmdBuffer.Length);
				cmdBuffer = newLargerBuffer;
			}

			Buffer.BlockCopy(cmdBytes, 0, cmdBuffer, cmdBufferIndex, cmdBytes.Length);
			cmdBufferIndex += cmdBytes.Length;
		}

		public void FlushSendBuffer()
		{
			socket.Send(cmdBuffer, cmdBufferIndex, SocketFlags.None);
			cmdBufferIndex = 0;
		}
        /// <summary>
        /// reset buffer index in send buffer
        /// </summary>
        public void ResetSendBuffer()
        {
            cmdBufferIndex = 0;
        }

		private int SafeReadByte()
		{
			return Bstream.ReadByte();
		}

		private void SendExpectSuccess(params byte[][] cmdWithBinaryArgs)
		{
			if (!SendCommand(cmdWithBinaryArgs))
				throw CreateConnectionError();

            if (Pipeline != null)
            {
                Pipeline.CompleteVoidQueuedCommand(ExpectSuccess);
                return;
            }
			ExpectSuccess();
		}

		private int SendExpectInt(params byte[][] cmdWithBinaryArgs)
		{
			if (!SendCommand(cmdWithBinaryArgs))
				throw CreateConnectionError();

            if (Pipeline != null)
            {
                Pipeline.CompleteIntQueuedCommand(ReadInt);
                return default(int);<commit_msg>Memory management of large byte [] in buffers<commit_after>//
// redis-sharp.cs: ECMA CLI Binding to the Redis key-value storage system
//
// Authors:
//   <NAME> (<EMAIL>)
//
// Copyright 2010 Novell, Inc.
//
// Licensed under the same terms of Redis: new BSD license.
//

using System;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using ServiceStack.Text;

namespace ServiceStack.Redis
{
	public partial class RedisNativeClient
	{
		private void Connect()
		{
			socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)
			{
				SendTimeout = SendTimeout,
				ReceiveTimeout = ReceiveTimeout
			};
			try
			{
                if (ConnectTimeout == 0)
                {
                    socket.Connect(Host, Port);
                }
                else
                {
                    var connectResult = socket.BeginConnect(Host, Port, null, null);
                    connectResult.AsyncWaitHandle.WaitOne(ConnectTimeout, true);
                }

				if (!socket.Connected)
				{
					socket.Close();
					socket = null;
					return;
				}
				Bstream = new BufferedStream(new NetworkStream(socket), 16 * 1024);

				if (Password != null)
					SendExpectSuccess(Commands.Auth, Password.ToUtf8Bytes());

				db = 0;
				var ipEndpoint = socket.LocalEndPoint as IPEndPoint;
				clientPort = ipEndpoint != null ? ipEndpoint.Port : -1;
				lastCommand = null;
				lastSocketException = null;
				LastConnectedAtTimestamp = Stopwatch.GetTimestamp();

                if (ConnectionFilter != null)
                {
                    ConnectionFilter(this);
                }
			}
			catch (SocketException ex)
			{
                if (socket != null)
                    socket.Close();
                socket = null;

				HadExceptions = true;
				var throwEx = new RedisException("could not connect to redis Instance at " + Host + ":" + Port, ex);
				log.Error(throwEx.Message, ex);
				throw throwEx;
			}
		}

		protected string ReadLine()
		{
			var sb = new StringBuilder();

			int c;
			while ((c = Bstream.ReadByte()) != -1)
			{
				if (c == '\r')
					continue;
				if (c == '\n')
					break;
				sb.Append((char)c);
			}
			return sb.ToString();
		}

		private bool AssertConnectedSocket()
		{
			if (LastConnectedAtTimestamp > 0)
			{
				var now = Stopwatch.GetTimestamp();
				var elapsedSecs = (now - LastConnectedAtTimestamp) / Stopwatch.Frequency;

				if (socket == null || (elapsedSecs > IdleTimeOutSecs && !socket.IsConnected()))
				{
					return Reconnect();
				}
				LastConnectedAtTimestamp = now;
			}

			if (socket == null)
			{
				var previousDb = db;
				Connect();
				if (previousDb != DefaultDb) this.Db = previousDb;
			}

			var isConnected = socket != null;

			return isConnected;
		}

		private bool Reconnect()
		{
			var previousDb = db;

			SafeConnectionClose();
			Connect(); //sets db to 0

			if (previousDb != DefaultDb) this.Db = previousDb;

			return socket != null;
		}

		private bool HandleSocketException(SocketException ex)
		{
			HadExceptions = true;
			log.Error("SocketException: ", ex);

			lastSocketException = ex;

			// timeout?
			socket.Close();
			socket = null;

			return false;
		}

		private RedisResponseException CreateResponseError(string error)
		{
			HadExceptions = true;
			var throwEx = new RedisResponseException(
				string.Format("{0}, sPort: {1}, LastCommand: {2}",
					error, clientPort, lastCommand));
			log.Error(throwEx.Message);
			throw throwEx;
		}

		private RedisException CreateConnectionError()
		{
			HadExceptions = true;
            var throwEx = new RedisException(
				string.Format("Unable to Connect: sPort: {0}",
					clientPort), lastSocketException);
			log.Error(throwEx.Message);
			throw throwEx;
		}

		private static byte[] GetCmdBytes(char cmdPrefix, int noOfLines)
		{
			var strLines = noOfLines.ToString();
			var strLinesLength = strLines.Length;

			var cmdBytes = new byte[1 + strLinesLength + 2];
			cmdBytes[0] = (byte)cmdPrefix;

			for (var i = 0; i < strLinesLength; i++)
				cmdBytes[i + 1] = (byte)strLines[i];

			cmdBytes[1 + strLinesLength] = 0x0D; // \r
			cmdBytes[2 + strLinesLength] = 0x0A; // \n

			return cmdBytes;
		}

		/// <summary>
		/// Command to set multuple binary safe arguments
		/// </summary>
		/// <param name="cmdWithBinaryArgs"></param>
		/// <returns></returns>
		protected bool SendCommand(params byte[][] cmdWithBinaryArgs)
		{
			if (!AssertConnectedSocket()) return false;

			try
			{
				CmdLog(cmdWithBinaryArgs);

				//Total command lines count
				WriteAllToSendBuffer(cmdWithBinaryArgs);

                //pipeline will handle flush, if pipelining is turned on
                if (Pipeline == null)
                    FlushSendBuffer();
			}
			catch (SocketException ex)
			{
                cmdBuffer.Clear();
				return HandleSocketException(ex);
			}
			return true;
		}

		public void WriteAllToSendBuffer(params byte[][] cmdWithBinaryArgs)
		{
			WriteToSendBuffer(GetCmdBytes('*', cmdWithBinaryArgs.Length));

			foreach (var safeBinaryValue in cmdWithBinaryArgs)
			{
				WriteToSendBuffer(GetCmdBytes('$', safeBinaryValue.Length));
				WriteToSendBuffer(safeBinaryValue);
				WriteToSendBuffer(endData);
			}
		}

        System.Collections.Generic.IList<ArraySegment<byte>> cmdBuffer = new System.Collections.Generic.List<ArraySegment<byte>>();

		public void WriteToSendBuffer(byte[] cmdBytes)
		{
            var copyOfBytes = new byte[cmdBytes.Length];
            Buffer.BlockCopy(cmdBytes, 0, copyOfBytes, 0, cmdBytes.Length);
            cmdBuffer.Add(new ArraySegment<byte>(copyOfBytes));
        }

		public void FlushSendBuffer()
		{
			socket.Send(cmdBuffer, SocketFlags.None);
            cmdBuffer.Clear();
		}
        /// <summary>
        /// reset buffer index in send buffer
        /// </summary>
        public void ResetSendBuffer()
        {
            cmdBuffer.Clear();
        }

		private int SafeReadByte()
		{
			return Bstream.ReadByte();
		}

		private void SendExpectSuccess(params byte[][] cmdWithBinaryArgs)
		{
			if (!SendCommand(cmdWithBinaryArgs))
				throw CreateConnectionError();

            if (Pipeline != null)
            {
                Pipeline.CompleteVoidQueuedCommand(ExpectSuccess);
                return;
            }
			ExpectSuccess();
		}

		private int SendExpectInt(params byte[][] cmdWithBinaryArgs)
		{
			if (!SendCommand(cmdWithBinaryArgs))
				throw CreateConnectionError();

            if (Pipeline != null)
            {
                Pipeline.CompleteIntQueuedCommand(ReadInt);
                return default(int);
{'fraction_non_alphanumeric': 0.11127895266868076, 'fraction_numerical': 0.006545820745216516, 'mean_word_length': 2.5313690627381256, 'pattern_counts': {'":': 2, '<': 32, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '3011840', 'n_tokens_mistral': 4999, 'n_tokens_neox': 4038, 'n_words': 983}