static int read_check(struct io_args *a, long n, const char *msg, int io_wait)
{
	if (n == -1) {
		if (errno == EINTR) {
			a->fd = my_fileno(a->io);
			return -1;
		}
		rb_str_set_len(a->buf, 0);
		if (errno == EAGAIN) {
			if (io_wait) {
				(void)kgio_call_wait_readable(a->io);

				/* buf may be modified in other thread/fiber */
				rb_str_modify(a->buf);
				rb_str_resize(a->buf, a->len);
				a->ptr = RSTRING_PTR(a->buf);
				return -1;
			} else {
				a->buf = sym_wait_readable;
				return 0;
			}
		}
		rd_sys_fail(msg);
	}
	rb_str_set_len(a->buf, n);
	if (n == 0)
		a->buf = Qnil;
	return 0;
}

/**
Gets returns a line. this is okay for small lines,
but shouldn't really be used.

Limited to ~ 1Mb of a line length.
*/
static VALUE tfio_gets(VALUE self) {
  int fd = get_tmpfile(self);
  size_t pos = get_pos(self);
  size_t end = get_end(self);
  if (pos == end)
    return Qnil;
  size_t pos_e = pos;
  char c;
  int ret;
  VALUE buffer;

  do {
    ret = pread(fd, &c, 1, pos_e);
  } while (ret > 0 && c != '\n' && (++pos_e < end));
  set_pos(self, pos_e + 1);
  if (pos > pos_e) {
    buffer = rb_str_buf_new(pos_e - pos);
    // make sure the buffer is binary encoded.
    rb_enc_associate(buffer, BinaryEncoding);
    if (pread(fd, RSTRING_PTR(buffer), pos_e - pos, pos) < 0)
      return Qnil;
    rb_str_set_len(buffer, pos_e - pos);
    return buffer;
  }
  return Qnil;
}

/*
 * call-seq:
 *      remove_scope_id(ip_address)
 *
 * Returns copy of IP address with Scope ID removed,
 * if address has it (only IPv6 actually may have it).
 */
static VALUE remove_scope_id(const char *addr)
{
	VALUE rv = rb_str_new2(addr);
	long len = RSTRING_LEN(rv);
	char *ptr = RSTRING_PTR(rv);
	char *pct = memchr(ptr, '%', len);

	/*
	 * remove scoped portion
	 * Ruby equivalent: rv.sub!(/%([^\]]*)\]/, "]")
	 */
	if (pct) {
		size_t newlen = pct - ptr;
		char *rbracket = memchr(pct, ']', len - newlen);

		if (rbracket) {
			size_t move = len - (rbracket - ptr);

			memmove(pct, rbracket, move);
			newlen += move;

			rb_str_set_len(rv, newlen);
		} else {
			rb_raise(rb_eArgError,
				"']' not found in IPv6 addr=%s", ptr);
                }
        }
        return rv;
}

/*
 *  call-seq:
 *     cipher.update(data [, buffer]) -> string or buffer
 *
 *  === Parameters
 *  +data+ is a nonempty string.
 *  +buffer+ is an optional string to store the result.
 */
static VALUE 
ossl_cipher_update(int argc, VALUE *argv, VALUE self)
{
    EVP_CIPHER_CTX *ctx;
    char *in;
    int in_len, out_len;
    VALUE data, str;

    rb_scan_args(argc, argv, "11", &data, &str);

    StringValue(data);
    in = RSTRING_PTR(data);
    if ((in_len = RSTRING_LEN(data)) == 0)
        rb_raise(rb_eArgError, "data must not be empty");
    GetCipher(self, ctx);
    out_len = in_len+EVP_CIPHER_CTX_block_size(ctx);

    if (NIL_P(str)) {
        str = rb_str_new(0, out_len);
    } else {
        StringValue(str);
        rb_str_resize(str, out_len);
    }

    if (!EVP_CipherUpdate(ctx, RSTRING_PTR(str), &out_len, in, in_len))
	ossl_raise(eCipherError, NULL);
    assert(out_len < RSTRING_LEN(str));
    rb_str_set_len(str, out_len);

    return str;
}

static VALUE in_addr_set(VALUE io, struct sockaddr_storage *addr, socklen_t len)
{
	VALUE host;
	int host_len, rc;
	char *host_ptr;

	switch (addr->ss_family) {
	case AF_INET:
		host_len = (long)INET_ADDRSTRLEN;
		break;
	case AF_INET6:
		host_len = (long)INET6_ADDRSTRLEN;
		break;
	default:
		rb_raise(rb_eRuntimeError, "unsupported address family");
	}
	host = rb_str_new(NULL, host_len);
	host_ptr = RSTRING_PTR(host);
	rc = getnameinfo((struct sockaddr *)addr, len,
			 host_ptr, host_len, NULL, 0, NI_NUMERICHOST);
	if (rc != 0)
		rb_raise(rb_eRuntimeError, "getnameinfo: %s", gai_strerror(rc));
	rb_str_set_len(host, strlen(host_ptr));
	return rb_ivar_set(io, iv_kgio_addr, host);
}

/*
 *  call-seq:
 *     cipher.update(data [, buffer]) -> string or buffer
 *
 *  Encrypts data in a streaming fashion. Hand consecutive blocks of data
 *  to the +update+ method in order to encrypt it. Returns the encrypted
 *  data chunk. When done, the output of Cipher#final should be additionally
 *  added to the result.
 *
 *  === Parameters
 *  +data+ is a nonempty string.
 *  +buffer+ is an optional string to store the result.
 */
static VALUE
ossl_cipher_update(int argc, VALUE *argv, VALUE self)
{
    EVP_CIPHER_CTX *ctx;
    unsigned char *in;
    long in_len, out_len;
    VALUE data, str;

    rb_scan_args(argc, argv, "11", &data, &str);

    StringValue(data);
    in = (unsigned char *)RSTRING_PTR(data);
    if ((in_len = RSTRING_LEN(data)) == 0)
        ossl_raise(rb_eArgError, "data must not be empty");
    GetCipher(self, ctx);
    out_len = in_len+EVP_CIPHER_CTX_block_size(ctx);
    if (out_len <= 0) {
	ossl_raise(rb_eRangeError,
		   "data too big to make output buffer: %ld bytes", in_len);
    }

    if (NIL_P(str)) {
        str = rb_str_new(0, out_len);
    } else {
        StringValue(str);
        rb_str_resize(str, out_len);
    }

    if (!ossl_cipher_update_long(ctx, (unsigned char *)RSTRING_PTR(str), &out_len, in, in_len))
	ossl_raise(eCipherError, NULL);
    assert(out_len < RSTRING_LEN(str));
    rb_str_set_len(str, out_len);

    return str;
}

static VALUE
rg_read_all(int argc, VALUE *argv, VALUE self)
{
        VALUE rbcount, cancellable, result;
        gsize count;
        GError *error = NULL;
        gsize bytes_read;

        rb_scan_args(argc, argv, "11", &rbcount, &cancellable);
        count = RVAL2GSIZE(rbcount);
        result = rb_str_new(NULL, count);
        if (!g_input_stream_read_all(_SELF(self),
                                     RSTRING_PTR(result),
                                     count,
                                     &bytes_read,
                                     RVAL2GCANCELLABLE(cancellable),
                                     &error))
                rbgio_raise_error(error);

        rb_str_set_len(result, bytes_read);
        rb_str_resize(result, bytes_read);
        OBJ_TAINT(result);

        return result;
}

static int
ruby__sfvwrite(register rb_printf_buffer *fp, register struct __suio *uio)
{
    struct __siov *iov;
    VALUE result = (VALUE)fp->_bf._base;
    char *buf = (char*)fp->_p;
    size_t len, n;
    size_t blen = buf - RSTRING_PTR(result), bsiz = fp->_w;

    if (RBASIC(result)->klass) {
	rb_raise(rb_eRuntimeError, "rb_vsprintf reentered");
    }
    if ((len = uio->uio_resid) == 0)
	return 0;
    CHECK(len);
    buf += blen;
    fp->_w = bsiz;
    for (iov = uio->uio_iov; len > 0; ++iov) {
	MEMCPY(buf, iov->iov_base, char, n = iov->iov_len);
	buf += n;
	len -= n;
    }
    fp->_p = (unsigned char *)buf;
    rb_str_set_len(result, buf - RSTRING_PTR(result));
    return 0;
}

// This function is equivalent to rb_str_cat(), but unlike the real
// rb_str_cat(), it doesn't leak memory in some versions of Ruby.
// For more information, see:
//   https://bugs.ruby-lang.org/issues/11328
VALUE noleak_rb_str_cat(VALUE rb_str, const char *str, long len) {
    size_t oldlen = RSTRING_LEN(rb_str);
    rb_str_modify_expand(rb_str, len);
    char *p = RSTRING_PTR(rb_str);
    memcpy(p + oldlen, str, len);
    rb_str_set_len(rb_str, oldlen + len);
}

// Reads data from the IO, according to the Rack specifications for `#read`.
static VALUE tfio_read(int argc, VALUE *argv, VALUE self) {
  int fd = get_tmpfile(self);
  size_t pos = get_pos(self);
  size_t end = get_end(self);
  VALUE buffer = Qnil;
  char ret_nil = 0;
  ssize_t len = 0;
  // get the buffer object if given
  if (argc == 2) {
    Check_Type(argv[1], T_STRING);
    buffer = argv[1];
  }
  // get the length object, if given
  if (argc > 0 && argv[0] != Qnil) {
    Check_Type(argv[0], T_FIXNUM);
    len = FIX2LONG(argv[0]);
    if (len < 0)
      rb_raise(rb_eRangeError, "length should be bigger then 0.");
    ret_nil = 1;
  }
  // return if we're at the EOF.
  if (pos == end)
    goto no_data;
  // calculate length if it wasn't specified.
  if (len == 0) {
    // make sure we're not reading more then we have
    len = end - pos;
    // set position for future reads
    set_pos(self, end);
    if (len == 0)
      goto no_data;
  } else {
    // set position for future reads
    set_pos(self, pos + len);
  }
  // limit read to what we have
  if (len + pos > end)
    len = end - pos;
  // create the buffer if we don't have one.
  if (buffer == Qnil) {
    buffer = rb_str_buf_new(len);
    // make sure the buffer is binary encoded.
    rb_enc_associate(buffer, BinaryEncoding);
  } else {
    // make sure the buffer is binary encoded.
    rb_enc_associate(buffer, BinaryEncoding);
    if (rb_str_capacity(buffer) < len)
      rb_str_resize(buffer, len);
  }
  // read the data.
  if (pread(fd, RSTRING_PTR(buffer), len, pos) <= 0)
    goto no_data;
  rb_str_set_len(buffer, len);
  return buffer;
no_data:
  if (ret_nil)
    return Qnil;
  else
    return rb_str_buf_new(0);
}

/*
 * call-seq:
 *    ssl.sysread(length) => string
 *    ssl.sysread(length, buffer) => buffer
 *
 * === Parameters
 * * +length+ is a positive integer.
 * * +buffer+ is a string used to store the result.
 */
static VALUE
ossl_ssl_read(int argc, VALUE *argv, VALUE self)
{
  SSL *ssl;
  int ilen, nread = 0;
  VALUE len, str;

  rb_scan_args(argc, argv, "11", &len, &str);
  ilen = NUM2INT(len);

  if(NIL_P(str)) {
    str = rb_str_new(0, ilen);
  } else {
    StringValue(str);
    rb_str_modify(str);
    rb_str_resize(str, ilen);
  }

  if(ilen == 0) return str;

  Data_Get_Struct(self, SSL, ssl);
  int fd = rb_io_fd(ossl_ssl_get_io(self));

  if (ssl) {
    if(SSL_pending(ssl) <= 0)
      rb_thread_wait_fd(fd);
    for (;;) {
      nread = SSL_read(ssl, RSTRING_PTR(str), RSTRING_LEN(str));
      switch(ssl_get_error(ssl, nread)) {
      case SSL_ERROR_NONE:
        goto end;
      case SSL_ERROR_ZERO_RETURN:
        rb_eof_error();
      case SSL_ERROR_WANT_WRITE:
        rb_io_wait_writable(fd);
        continue;
      case SSL_ERROR_WANT_READ:
        rb_io_wait_readable(fd);
        continue;
      case SSL_ERROR_SYSCALL:
        if(ERR_peek_error() == 0 && nread == 0) rb_eof_error();
        rb_sys_fail(0);
      default:
        ossl_raise(eSSLError, "SSL_read:");
      }
    }
  }
  else {
    ID id_sysread = rb_intern("sysread");
    rb_warning("SSL session is not started yet.");
    return rb_funcall(ossl_ssl_get_io(self), id_sysread, 2, len, str);
  }

end:
  rb_str_set_len(str, nread);
  OBJ_TAINT(str);

  return str;
}

	/* Use somewhat faster version with access to string capacity on MRI */
	char *
	pg_rb_str_ensure_capa( VALUE str, long expand_len, char *curr_ptr, char **end_ptr )
	{
		long curr_len = curr_ptr - RSTRING_PTR(str);
		long curr_capa = rb_str_capacity( str );
		if( curr_capa < curr_len + expand_len ){
			rb_str_set_len( str, curr_len );
			rb_str_modify_expand( str, (curr_len + expand_len) * 2 - curr_capa );
			curr_ptr = RSTRING_PTR(str) + curr_len;
		}
		if( end_ptr )
			*end_ptr = RSTRING_PTR(str) + rb_str_capacity( str );
		return curr_ptr;
	}

/*
 * call-seq:
 *    coder.encode( value [, encoding] )
 *
 * Encodes the given Ruby object into string representation, without
 * sending data to/from the database server.
 *
 * A nil value is passed through.
 *
 */
static VALUE
pg_coder_encode(int argc, VALUE *argv, VALUE self)
{
	VALUE res;
	VALUE intermediate;
	VALUE value;
	int len, len2;
	int enc_idx;
	t_pg_coder *this = DATA_PTR(self);

	if(argc < 1 || argc > 2){
		rb_raise(rb_eArgError, "wrong number of arguments (%i for 1..2)", argc);
	}else if(argc == 1){
		enc_idx = rb_ascii8bit_encindex();
	}else{
		enc_idx = rb_to_encoding_index(argv[1]);
	}
	value = argv[0];

	if( NIL_P(value) )
		return Qnil;

	if( !this->enc_func ){
		rb_raise(rb_eRuntimeError, "no encoder function defined");
	}

	len = this->enc_func( this, value, NULL, &intermediate, enc_idx );

	if( len == -1 ){
		/* The intermediate value is a String that can be used directly. */
		OBJ_INFECT(intermediate, value);
		return intermediate;
	}

	res = rb_str_new(NULL, len);
	PG_ENCODING_SET_NOCHECK(res, enc_idx);
	len2 = this->enc_func( this, value, RSTRING_PTR(res), &intermediate, enc_idx );
	if( len < len2 ){
		rb_bug("%s: result length of first encoder run (%i) is less than second run (%i)",
			rb_obj_classname( self ), len, len2 );
	}
	rb_str_set_len( res, len2 );
	OBJ_INFECT(res, value);

	RB_GC_GUARD(intermediate);

	return res;
}

/*
 *  call-seq:
 *     cipher.final -> string
 *
 *  Returns the remaining data held in the cipher object. Further calls to
 *  Cipher#update or Cipher#final will return garbage. This call should always
 *  be made as the last call of an encryption or decryption operation, after
 *  after having fed the entire plaintext or ciphertext to the Cipher instance.
 *
 *  If an authenticated cipher was used, a CipherError is raised if the tag
 *  could not be authenticated successfully. Only call this method after
 *  setting the authentication tag and passing the entire contents of the
 *  ciphertext into the cipher.
 */
static VALUE
ossl_cipher_final(VALUE self)
{
    EVP_CIPHER_CTX *ctx;
    int out_len;
    VALUE str;

    GetCipher(self, ctx);
    str = rb_str_new(0, EVP_CIPHER_CTX_block_size(ctx));
    if (!EVP_CipherFinal_ex(ctx, (unsigned char *)RSTRING_PTR(str), &out_len))
	ossl_raise(eCipherError, NULL);
    assert(out_len <= RSTRING_LEN(str));
    rb_str_set_len(str, out_len);

    return str;
}

static VALUE
rg_peek(VALUE self, VALUE rboffset, VALUE rbcount)
{
        gsize offset = RVAL2GSIZE(rboffset);
        gsize count = RVAL2GSIZE(rbcount);
        VALUE result = rb_str_new(NULL, count);
        gsize bytes_peeked = g_buffered_input_stream_peek(_SELF(self),
                                                          RSTRING_PTR(result),
                                                          offset,
                                                          count);

        rb_str_set_len(result, bytes_peeked);
        rb_str_resize(result, bytes_peeked);
        OBJ_TAINT(result);

        return result;
}

VALUE mCDB_Reader_get(VALUE self, VALUE key) {
  struct cdb db;
  VALUE value;
  StringValue(key);
  int fd = open_cdb_fd(self);
  size_t vlen;
  cdb_init(&db,fd);
  if(cdb_find(&db,RSTRING_PTR(key),RSTRING_LEN(key)) > 0) {
    vlen = cdb_datalen(&db);
    value = rb_str_buf_new(vlen);
    cdb_read(&db,RSTRING_PTR(value),vlen,cdb_datapos(&db));
    rb_str_set_len(value,vlen);
    return value;
  }
  close(fd);
  return Qnil;
}

static VALUE
pkcs11_C_GetOperationState(VALUE self, VALUE session)
{
  CK_RV rv;
  CK_C_GetOperationState func;
  VALUE state;
  CK_ULONG size;

  GetFunction(self, C_GetOperationState, func);
  CallFunction(C_GetOperationState, func, rv, NUM2HANDLE(session), NULL_PTR, &size);
  if (rv != CKR_OK) pkcs11_raise(self,rv);
  state = rb_str_new(0, size);
  CallFunction(C_GetOperationState, func, rv, NUM2HANDLE(session), (CK_BYTE_PTR)RSTRING_PTR(state), &size);
  if (rv != CKR_OK) pkcs11_raise(self,rv);
  rb_str_set_len(state, size);

  return state;
}

VALUE mCDB_Reader_each_for_key(VALUE self,VALUE key) {
  struct cdb db;
  VALUE value;
  struct cdb_find find;
  StringValue(key);
  int fd = open_cdb_fd(self);
  size_t vlen;
  cdb_init(&db,fd);
  cdb_findinit(&find,&db,RSTRING_PTR(key),RSTRING_LEN(key));
  while(cdb_findnext(&find) > 0) {
    vlen = cdb_datalen(&db);
    value = rb_str_buf_new(vlen);
    cdb_read(&db,RSTRING_PTR(value),vlen,cdb_datapos(&db));
    rb_str_set_len(value,vlen);
    rb_yield(value);
  }
  close(fd);
  return Qnil;
}

static VALUE _receive(int rflags, int argc, VALUE *argv, VALUE self)
{
	struct posix_mq *mq = get(self, 1);
	struct rw_args x;
	VALUE buffer, timeout;
	struct timespec expire;

	if (mq->attr.mq_msgsize < 0) {
		if (mq_getattr(mq->des, &mq->attr) < 0)
			rb_sys_fail("mq_getattr");
	}

	rb_scan_args(argc, argv, "02", &buffer, &timeout);
	x.timeout = convert_timeout(&expire, timeout);

	if (NIL_P(buffer)) {
		buffer = rb_str_new(0, mq->attr.mq_msgsize);
	} else {
		StringValue(buffer);
		rb_str_modify(buffer);
		rb_str_resize(buffer, mq->attr.mq_msgsize);
	}
	OBJ_TAINT(buffer);
	x.msg_ptr = RSTRING_PTR(buffer);
	x.msg_len = (size_t)mq->attr.mq_msgsize;
	x.des = mq->des;

retry:
	WITHOUT_GVL(xrecv, &x, RUBY_UBF_IO, 0);
	if (x.received < 0) {
		if (errno == EINTR)
			goto retry;
		if (errno == EAGAIN && (rflags & PMQ_TRY))
			return Qnil;
		rb_sys_fail("mq_receive");
	}

	rb_str_set_len(buffer, x.received);

	if (rflags & PMQ_WANTARRAY)
		return rb_ary_new3(2, buffer, UINT2NUM(x.msg_prio));
	return buffer;
}

void
rb_sys_enc_warning(rb_encoding *enc, const char *fmt, ...)
{
    VALUE mesg;
    va_list args;
    int errno_save;

    errno_save = errno;

    if (!RTEST(ruby_verbose)) return;

    va_start(args, fmt);
    mesg = warning_string(enc, fmt, args);
    va_end(args);
    rb_str_set_len(mesg, RSTRING_LEN(mesg)-1);
    rb_str_catf(mesg, ": %s\n", strerror(errno_save));
    rb_write_error_str(mesg);
    errno = errno_save;
}